From da6aded7b50e430cc9fbf0c0c78750167f3ff8a6 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Sun, 19 Jul 2026 15:06:50 +0800 Subject: [PATCH] [core][spark] Support catalog-managed Format Table reads and writes Route managed Format Table partition discovery through the catalog so scans read exactly the registered partitions and an empty registration reads as empty instead of falling back to directory listing. Validate partition locations against the table path before any read or delete. Register written partitions at commit time in bounded batches, keep IF NOT EXISTS registration idempotent for concurrent writers, and fall back to unmanaged directory discovery when no catalog manages the table. --- docs/docs/flink/sql-ddl.md | 16 + docs/generated/core_configuration.html | 3 +- .../java/org/apache/paimon/CoreOptions.java | 5 +- .../apache/paimon/catalog/CachingCatalog.java | 30 +- .../org/apache/paimon/catalog/Catalog.java | 11 + .../apache/paimon/catalog/CatalogUtils.java | 102 +++- .../paimon/catalog/DelegateCatalog.java | 5 + .../org/apache/paimon/rest/RESTCatalog.java | 29 +- .../org/apache/paimon/table/FormatTable.java | 79 ++- .../table/format/FormatBatchWriteBuilder.java | 3 +- .../table/format/FormatReadBuilder.java | 10 + .../format/FormatTableCatalogProvider.java | 191 ++++++ .../table/format/FormatTableCommit.java | 148 ++++- .../paimon/table/format/FormatTableScan.java | 25 +- .../table/format/ManagedFormatTableScan.java | 209 +++++++ .../paimon/utils/PartitionPathUtils.java | 160 ++++- .../paimon/catalog/CachingCatalogTest.java | 26 + .../paimon/catalog/CatalogUtilsTest.java | 131 +++++ .../paimon/catalog/FileSystemCatalogTest.java | 52 ++ .../paimon/rest/MockRESTCatalogTest.java | 94 +++ .../apache/paimon/rest/RESTCatalogServer.java | 40 ++ .../apache/paimon/rest/RESTCatalogTest.java | 58 ++ .../table/FormatTableCompatibilityTest.java | 185 ++++++ .../table/format/FormatTableCommitTest.java | 247 ++++++++ .../format/ManagedFormatTableScanTest.java | 553 ++++++++++++++++++ .../paimon/utils/PartitionPathUtilsTest.java | 50 +- .../format/FormatTableBatchWriteBase.scala | 9 + .../format/FormatTableBatchWriteTest.scala | 98 ++++ 28 files changed, 2517 insertions(+), 52 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala diff --git a/docs/docs/flink/sql-ddl.md b/docs/docs/flink/sql-ddl.md index 4964a0cc18d2..ff5dea32699e 100644 --- a/docs/docs/flink/sql-ddl.md +++ b/docs/docs/flink/sql-ddl.md @@ -34,6 +34,22 @@ Paimon catalogs currently support three types of metastores: See [CatalogOptions](../maintenance/configurations#catalogoptions) for detailed options when creating a catalog. +:::info + +For Format Tables, `metastore.partitioned-table = true` enables catalog-managed partitions, which +requires an internal table in a catalog that supports it (currently the REST catalog) and cannot +be combined with `format-table.implementation = engine`. The REST catalog validates this +combination on `CREATE TABLE` (catalog-level `table-default.*` options participate in the +effective options); other catalogs keep their previous behavior and treat the option as inert. + +A Format Table that carries `metastore.partitioned-table = true` where it cannot be honored +(e.g. in a Hive catalog, or on an external table) still loads — the option is ignored, the table +behaves as an unmanaged Format Table, and a warning is logged. To remove the option permanently, +use `ALTER TABLE my_table RESET ('metastore.partitioned-table')` where the catalog supports +altering the table. + +::: + ### Create Filesystem Catalog The following Flink SQL registers and uses a Paimon catalog named `my_catalog`. Metadata and table files are stored under `hdfs:///path/to/warehouse`. diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 143c244451a9..f4e4098dd263 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1029,7 +1029,8 @@ Boolean Whether to create this table as a partitioned table in metastore. For example, if you want to list all partitions of a Paimon table in Hive, you need to create this table as a partitioned table in Hive metastore. -This config option does not affect the default filesystem metastore. +This config option does not affect the default filesystem metastore. +For format tables in a REST catalog, setting this option makes partition metadata catalog-managed: partitions must be registered in the catalog, and partition listing does not fall back to filesystem discovery.
metastore.tag-to-partition
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 4f1cb3111193..96964c7fe72c 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -1823,7 +1823,10 @@ public String toString() { "Whether to create this table as a partitioned table in metastore.\n" + "For example, if you want to list all partitions of a Paimon table in Hive, " + "you need to create this table as a partitioned table in Hive metastore.\n" - + "This config option does not affect the default filesystem metastore."); + + "This config option does not affect the default filesystem metastore.\n" + + "For format tables in a REST catalog, setting this option makes partition " + + "metadata catalog-managed: partitions must be registered in the catalog, " + + "and partition listing does not fall back to filesystem discovery."); public static final ConfigOption METASTORE_TAG_TO_PARTITION = key("metastore.tag-to-partition") diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java index 01f04cbba3ae..1b7ac137e036 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CachingCatalog.java @@ -338,9 +338,10 @@ public List listPartitions(Identifier identifier) throws TableNotExis @Override public void createPartitions(Identifier identifier, List> partitions) throws TableNotExistException { - wrapped.createPartitions(identifier, partitions); - if (partitionCache != null) { - partitionCache.invalidate(identifier); + try { + wrapped.createPartitions(identifier, partitions); + } finally { + invalidatePartitionCache(identifier); } } @@ -348,25 +349,34 @@ public void createPartitions(Identifier identifier, List> pa public void createPartitions( Identifier identifier, List> partitions, boolean ignoreIfExists) throws TableNotExistException { - wrapped.createPartitions(identifier, partitions, ignoreIfExists); - if (partitionCache != null) { - partitionCache.invalidate(identifier); + try { + wrapped.createPartitions(identifier, partitions, ignoreIfExists); + } finally { + invalidatePartitionCache(identifier); } } @Override public void dropPartitions(Identifier identifier, List> partitions) throws TableNotExistException { - wrapped.dropPartitions(identifier, partitions); - if (partitionCache != null) { - partitionCache.invalidate(identifier); + try { + wrapped.dropPartitions(identifier, partitions); + } finally { + invalidatePartitionCache(identifier); } } @Override public void alterPartitions(Identifier identifier, List partitions) throws TableNotExistException { - wrapped.alterPartitions(identifier, partitions); + try { + wrapped.alterPartitions(identifier, partitions); + } finally { + invalidatePartitionCache(identifier); + } + } + + private void invalidatePartitionCache(Identifier identifier) { if (partitionCache != null) { partitionCache.invalidate(identifier); } diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java index 8e4ad8d51815..8693c8a259a5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java @@ -670,6 +670,17 @@ default boolean supportsListTableByType() { return false; } + /** + * Whether this catalog can be the source of truth for partitions of managed Format Tables + * ({@code type=format-table} with {@code metastore.partitioned-table=true}), i.e. whether it + * implements the partition APIs that managed Format Table scans and commits rely on. Catalogs + * without this capability load such tables with the option ignored, falling back to filesystem + * partition discovery. + */ + default boolean supportsManagedFormatTablePartitions() { + return false; + } + // ==================== Version management methods ========================== /** diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java index 0c34fc39b1d8..c8a4780d7f4c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java @@ -39,6 +39,7 @@ import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.Table; import org.apache.paimon.table.TableSnapshot; +import org.apache.paimon.table.format.FormatTableCatalogProvider; import org.apache.paimon.table.iceberg.IcebergTable; import org.apache.paimon.table.lance.LanceTable; import org.apache.paimon.table.object.ObjectTable; @@ -54,6 +55,9 @@ import org.apache.paimon.utils.Pair; import org.apache.paimon.utils.Preconditions; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import javax.annotation.Nullable; import java.util.ArrayList; @@ -85,6 +89,8 @@ /** Utils for {@link Catalog}. */ public class CatalogUtils { + private static final Logger LOG = LoggerFactory.getLogger(CatalogUtils.class); + public static Path path(String warehouse, String database, String table) { return new Path(String.format("%s/%s.db/%s", warehouse, database, table)); } @@ -173,6 +179,7 @@ private static void validateFormatTableOptions(Options options, boolean dataToke options.get(PRIMARY_KEY) == null, "Cannot define %s for format table.", PRIMARY_KEY.key()); + validateManagedFormatTableOptions(options); if (dataTokenEnabled && options.get(PATH) == null) { checkArgument( options.get(FORMAT_TABLE_IMPLEMENTATION) @@ -213,6 +220,44 @@ private static void validateFormatTableOptions(Options options, boolean dataToke } } + /** Validate options which are specific to catalog-managed format tables. */ + public static void validateManagedFormatTableOptions(Map tableOptions) { + validateManagedFormatTableOptions(Options.fromMap(tableOptions)); + } + + private static void validateManagedFormatTableOptions(Options options) { + if (options.get(CoreOptions.METASTORE_PARTITIONED_TABLE)) { + checkArgument( + options.get(FORMAT_TABLE_IMPLEMENTATION) + != CoreOptions.FormatTableImplementation.ENGINE, + "Cannot define %s=true together with %s=engine for a managed format table.", + CoreOptions.METASTORE_PARTITIONED_TABLE.key(), + FORMAT_TABLE_IMPLEMENTATION.key()); + } + } + + /** Validate that a catalog can persist the requested managed Format Table. */ + public static void validateManagedFormatTableCatalog( + Identifier identifier, + Map tableOptions, + Catalog catalog, + boolean isExternal) { + CoreOptions options = CoreOptions.fromMap(tableOptions); + if (options.type() != TableType.FORMAT_TABLE || !options.partitionedTableInMetastore()) { + return; + } + checkArgument( + !isExternal, + "Managed format table %s must be an internal table.", + identifier.getFullName()); + checkArgument( + catalog.supportsManagedFormatTablePartitions(), + "Managed format table %s requires a catalog with catalog-managed partition support" + + " (e.g. a REST catalog), but %s does not support it.", + identifier.getFullName(), + catalog.getClass().getSimpleName()); + } + public static void validateNamePattern(Catalog catalog, String namePattern) { if (Objects.nonNull(namePattern) && !catalog.supportsListByPattern()) { throw new UnsupportedOperationException( @@ -271,6 +316,34 @@ public static List listPartitionsFromFileSystem( return partitions; } + /** + * Why the managed Format Table semantics cannot be honored when loading the table from this + * catalog, or {@code null} if they can. + */ + @Nullable + private static String managedFormatTableLoadProblem( + Catalog catalog, TableMetadata metadata, Map tableOptions) { + if (!catalog.supportsManagedFormatTablePartitions()) { + return String.format( + "catalog %s does not support catalog-managed Format Table partitions", + catalog.getClass().getSimpleName()); + } + if (metadata.isExternal()) { + return "managed format tables must be internal tables"; + } + if (Options.fromMap(tableOptions).get(FORMAT_TABLE_IMPLEMENTATION) + == CoreOptions.FormatTableImplementation.ENGINE) { + return String.format( + "the option conflicts with %s=engine", FORMAT_TABLE_IMPLEMENTATION.key()); + } + if (catalog.catalogLoader() == null) { + return String.format( + "catalog %s does not provide a catalog loader", + catalog.getClass().getSimpleName()); + } + return null; + } + /** * Load table from {@link Catalog}, this table can be: * @@ -304,7 +377,30 @@ public static Table loadTable( Function dataFileIO = metadata.isExternal() ? externalFileIO : internalFileIO; if (options.type() == TableType.FORMAT_TABLE) { - return toFormatTable(identifier, schema, dataFileIO, catalogContext); + FormatTableCatalogProvider catalogProvider = null; + if (options.partitionedTableInMetastore()) { + String problem = managedFormatTableLoadProblem(catalog, metadata, schema.options()); + if (problem == null) { + catalogProvider = + new FormatTableCatalogProvider(identifier, catalog.catalogLoader()); + } else { + // Existing tables may carry the option from a version (or catalog) where it + // was inert; refusing to load them would leave no SQL way to repair the + // table. Keep them readable with filesystem partition discovery instead. + LOG.warn( + "Ignoring '{}=true' on format table {}: {}. Falling back to " + + "filesystem partition discovery, the table behaves as an " + + "unmanaged format table. Use ALTER TABLE ... RESET/UNSET " + + "to remove the option.", + CoreOptions.METASTORE_PARTITIONED_TABLE.key(), + identifier, + problem); + Map effectiveOptions = new HashMap<>(schema.options()); + effectiveOptions.remove(CoreOptions.METASTORE_PARTITIONED_TABLE.key()); + schema = schema.copy(effectiveOptions); + } + } + return toFormatTable(identifier, schema, dataFileIO, catalogContext, catalogProvider); } if (options.type() == TableType.OBJECT_TABLE) { @@ -453,7 +549,8 @@ private static FormatTable toFormatTable( Identifier identifier, TableSchema schema, Function fileIO, - CatalogContext catalogContext) { + CatalogContext catalogContext, + @Nullable FormatTableCatalogProvider catalogProvider) { Map options = schema.options(); FormatTable.Format format = FormatTable.parseFormat( @@ -471,6 +568,7 @@ private static FormatTable toFormatTable( .options(options) .comment(schema.comment()) .catalogContext(catalogContext) + .catalogProvider(catalogProvider) .build(); } diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java index 290849024fdb..8685fde9d7fc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java @@ -198,6 +198,11 @@ public boolean supportsListTableByType() { return wrapped.supportsListTableByType(); } + @Override + public boolean supportsManagedFormatTablePartitions() { + return wrapped.supportsManagedFormatTablePartitions(); + } + @Override public boolean supportsVersionManagement() { return wrapped.supportsVersionManagement(); diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java index c44378f6d00f..9f0737f868fe 100644 --- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java @@ -63,6 +63,7 @@ import org.apache.paimon.table.Instant; import org.apache.paimon.table.Table; import org.apache.paimon.table.TableSnapshot; +import org.apache.paimon.table.format.FormatTableCatalogProvider; import org.apache.paimon.table.sink.BatchTableCommit; import org.apache.paimon.table.system.SystemTableLoader; import org.apache.paimon.utils.Pair; @@ -95,6 +96,8 @@ import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase; import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem; import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable; +import static org.apache.paimon.catalog.CatalogUtils.validateManagedFormatTableCatalog; +import static org.apache.paimon.catalog.CatalogUtils.validateManagedFormatTableOptions; import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE; /** A catalog implementation for REST. */ @@ -453,6 +456,11 @@ public boolean supportsListTableByType() { return true; } + @Override + public boolean supportsManagedFormatTablePartitions() { + return true; + } + @Override public boolean supportsVersionManagement() { return true; @@ -567,8 +575,13 @@ public void createTable(Identifier identifier, Schema schema, boolean ignoreIfEx checkNotBranch(identifier, "createTable"); checkNotSystemTable(identifier, "createTable"); validateCreateTable(schema, dataTokenEnabled); - createExternalTablePathIfNotExist(schema); tableDefaultOptions.forEach(schema.options()::putIfAbsent); + // Defaults participate in the managed Format Table combination, so validate the + // effective options rather than only the explicit ones. + validateManagedFormatTableOptions(schema.options()); + validateManagedFormatTableCatalog( + identifier, schema.options(), this, schema.options().containsKey(PATH.key())); + createExternalTablePathIfNotExist(schema); Schema newSchema = inferSchemaIfExternalPaimonTable(schema); api.createTable(identifier, newSchema); } catch (AlreadyExistsException e) { @@ -645,8 +658,14 @@ public void replaceTable(Identifier identifier, Schema newSchema, boolean ignore checkNotBranch(identifier, "replaceTable"); checkNotSystemTable(identifier, "replaceTable"); validateCreateTable(newSchema, dataTokenEnabled); + tableDefaultOptions.forEach(newSchema.options()::putIfAbsent); + // Defaults participate in the managed Format Table combination, so validate the + // effective options rather than only the explicit ones. Externality is not decided + // client-side here: a round-tripped schema of an internal table may carry the synthetic + // path option, and the server remains the authority for replace semantics. + validateManagedFormatTableOptions(newSchema.options()); + validateManagedFormatTableCatalog(identifier, newSchema.options(), this, false); try { - tableDefaultOptions.forEach(newSchema.options()::putIfAbsent); api.replaceTable(identifier, newSchema); } catch (NoSuchResourceException e) { if (!ignoreIfNotExists) { @@ -751,6 +770,9 @@ public void createPartitions( identifier, e.getMessage())); } catch (BadRequestException e) { throw new IllegalArgumentException(e.getMessage()); + } finally { + // The server may have committed even when the response was lost or malformed. + FormatTableCatalogProvider.advanceGeneration(identifier); } } @@ -769,6 +791,9 @@ public void dropPartitions(Identifier identifier, List> part throw new TableNoPermissionException(identifier, e); } catch (BadRequestException e) { throw new IllegalArgumentException(e.getMessage()); + } finally { + // The server may have committed even when the response was lost or malformed. + FormatTableCatalogProvider.advanceGeneration(identifier); } return; } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java index 9a832176caaf..64325e70b7c8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java @@ -18,9 +18,11 @@ package org.apache.paimon.table; +import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; import org.apache.paimon.annotation.Public; import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.CatalogUtils; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.fs.FileIO; import org.apache.paimon.manifest.IndexManifestEntry; @@ -29,6 +31,7 @@ import org.apache.paimon.stats.Statistics; import org.apache.paimon.table.format.FormatBatchWriteBuilder; import org.apache.paimon.table.format.FormatReadBuilder; +import org.apache.paimon.table.format.FormatTableCatalogProvider; import org.apache.paimon.table.sink.BatchWriteBuilder; import org.apache.paimon.table.sink.StreamWriteBuilder; import org.apache.paimon.table.source.BatchVectorSearchBuilder; @@ -56,8 +59,9 @@ * operations on this table allow for reading or writing to these files, facilitating the retrieval * of existing data and the addition of new files. * - *

Partitioned file format table just like the standard hive format. Partitions are discovered - * and inferred based on directory structure. + *

A partitioned file format table uses the standard Hive directory layout. By default, + * partitions are discovered from that layout. Catalog-managed format tables use catalog metadata + * for partition visibility while retaining the same physical layout. * * @since 0.9.0 */ @@ -75,6 +79,12 @@ public interface FormatTable extends Table { CatalogContext catalogContext(); + /** Catalog access used by managed partition discovery and registration. */ + @Nullable + default FormatTableCatalogProvider catalogProvider() { + return null; + } + /** Currently supported formats. */ enum Format { ORC, @@ -115,6 +125,7 @@ class Builder { private Map options; @Nullable private String comment; private CatalogContext catalogContext; + @Nullable private FormatTableCatalogProvider catalogProvider; public Builder fileIO(FileIO fileIO) { this.fileIO = fileIO; @@ -161,6 +172,11 @@ public Builder catalogContext(CatalogContext catalogContext) { return this; } + public Builder catalogProvider(@Nullable FormatTableCatalogProvider catalogProvider) { + this.catalogProvider = catalogProvider; + return this; + } + public FormatTable build() { return new FormatTableImpl( fileIO, @@ -171,7 +187,8 @@ public FormatTable build() { format, options, comment, - catalogContext); + catalogContext, + catalogProvider); } } @@ -189,6 +206,7 @@ class FormatTableImpl implements FormatTable { private final Map options; @Nullable private final String comment; private CatalogContext catalogContext; + @Nullable private final FormatTableCatalogProvider catalogProvider; public FormatTableImpl( FileIO fileIO, @@ -200,6 +218,30 @@ public FormatTableImpl( Map options, @Nullable String comment, CatalogContext catalogContext) { + this( + fileIO, + identifier, + rowType, + partitionKeys, + location, + format, + options, + comment, + catalogContext, + null); + } + + public FormatTableImpl( + FileIO fileIO, + Identifier identifier, + RowType rowType, + List partitionKeys, + String location, + Format format, + Map options, + @Nullable String comment, + CatalogContext catalogContext, + @Nullable FormatTableCatalogProvider catalogProvider) { this.fileIO = fileIO; this.identifier = identifier; this.rowType = rowType; @@ -209,6 +251,7 @@ public FormatTableImpl( this.options = options; this.comment = comment; this.catalogContext = catalogContext; + this.catalogProvider = catalogProvider; } @Override @@ -265,6 +308,27 @@ public FileIO fileIO() { public FormatTable copy(Map dynamicOptions) { Map newOptions = new HashMap<>(options); newOptions.putAll(dynamicOptions); + + CoreOptions coreOptions = CoreOptions.fromMap(options); + CoreOptions copiedCoreOptions = CoreOptions.fromMap(newOptions); + boolean managed = coreOptions.partitionedTableInMetastore(); + boolean copiedManaged = copiedCoreOptions.partitionedTableInMetastore(); + if (managed != copiedManaged) { + throw new IllegalArgumentException( + String.format( + "Dynamic option '%s' cannot change whether Format Table partitions are catalog-managed.", + CoreOptions.METASTORE_PARTITIONED_TABLE.key())); + } + if (managed + && coreOptions.formatTablePartitionOnlyValueInPath() + != copiedCoreOptions.formatTablePartitionOnlyValueInPath()) { + throw new IllegalArgumentException( + String.format( + "Dynamic option '%s' cannot change the physical partition layout of a catalog-managed Format Table.", + CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key())); + } + CatalogUtils.validateManagedFormatTableOptions(newOptions); + return new FormatTableImpl( fileIO, identifier, @@ -274,7 +338,8 @@ public FormatTable copy(Map dynamicOptions) { format, newOptions, comment, - catalogContext); + catalogContext, + catalogProvider); } @Override @@ -302,6 +367,12 @@ public FullTextSearchBuilder newFullTextSearchBuilder() { public CatalogContext catalogContext() { return this.catalogContext; } + + @Override + @Nullable + public FormatTableCatalogProvider catalogProvider() { + return catalogProvider; + } } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java index 924e67dbdfdc..d3970bb878ea 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java @@ -87,7 +87,8 @@ public BatchTableCommit newCommit() { Identifier.fromString(table.fullName()), staticPartition, syncHiveUri, - table.catalogContext()); + table.catalogContext(), + table.catalogProvider()); } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java index 30a567b436a6..1c9010d0be04 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java @@ -173,6 +173,16 @@ public TableScan newScan() { partitionFilter = partitionPredicateOpt.get(); } } + if (options.partitionedTableInMetastore()) { + if (table.catalogProvider() == null) { + throw new IllegalStateException( + String.format( + "Managed format table %s has no catalog partition provider. " + + "The catalog client does not support managed format tables.", + table.fullName())); + } + return new ManagedFormatTableScan(table, partitionFilter, limit); + } return new FormatTableScan(table, partitionFilter, limit); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java new file mode 100644 index 000000000000..c297ef63e094 --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java @@ -0,0 +1,191 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.PagedList; +import org.apache.paimon.annotation.Experimental; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.CatalogLoader; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.utils.StringUtils; + +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; + +/** Serializable catalog access for a managed format table. */ +@Experimental +public class FormatTableCatalogProvider implements Serializable { + + private static final long serialVersionUID = 1L; + + private static final int PARTITION_PAGE_SIZE = 1000; + private static final int MAX_CACHED_PATTERNS = 128; + private static final int MAX_TRACKED_TABLE_GENERATIONS = 10_000; + private static final Duration PARTITION_CACHE_TTL = Duration.ofSeconds(30); + private static final Duration TABLE_GENERATION_TTL = Duration.ofHours(1); + + // Keyed by identifier full name: listings themselves live in per-instance caches, so sharing + // a generation counter across table incarnations (or across catalogs using the same name) can + // only cause an extra invalidation, never a stale read. + private static final Cache GENERATIONS = + Caffeine.newBuilder() + .expireAfterAccess(TABLE_GENERATION_TTL) + .maximumSize(MAX_TRACKED_TABLE_GENERATIONS) + .executor(Runnable::run) + .build(); + + private final Identifier identifier; + private final CatalogLoader catalogLoader; + + @Nullable private transient Cache> partitionCache; + // Reused across list/create calls: constructing a Catalog (HTTP client, auth provider, and + // for some auth providers an initial token fetch) per partition operation is expensive, and + // RESTCatalog.close() is a no-op so a long-lived instance leaks nothing. Recreated lazily + // after deserialization. + @Nullable private transient Catalog catalog; + + public FormatTableCatalogProvider(Identifier identifier, CatalogLoader catalogLoader) { + this.identifier = identifier; + this.catalogLoader = catalogLoader; + } + + /** List every catalog-visible partition matching the optional name prefix pattern. */ + public List listPartitions(@Nullable String partitionNamePattern) { + return cache().get( + partitionCacheKey(partitionNamePattern), + ignored -> loadPartitions(partitionNamePattern)); + } + + /** + * Advance the partition-listing generation of every provider of this table in this JVM, so + * same-process scans skip cached listings taken before the mutation. The catalog partition DDL + * path (create/dropPartitions) calls this from a {@code finally} block: the mutation's outcome + * can be ambiguous (the server may have committed even when the response was lost), so the + * cached listing must be dropped after every attempt, not only after a confirmed success. + */ + public static void advanceGeneration(Identifier identifier) { + GENERATIONS.get(identifier.getFullName(), ignored -> new AtomicLong()).incrementAndGet(); + } + + /** Create partitions with the catalog's idempotent create contract. */ + public void createPartitions(List> partitions) { + if (partitions.isEmpty()) { + return; + } + try { + // Register in bounded batches: one backfill commit can touch tens of thousands of + // partitions (e.g. years of hourly data in a single dynamic-partition INSERT), and a + // single unbounded request either monopolizes a catalog worker or trips server-side + // request caps, failing the job after its data files are already committed. + // Registration is an idempotent ADD-only upsert, so a mid-way failure leaves a state + // that a rerun or MSCK converges from. + for (int start = 0; start < partitions.size(); start += PARTITION_PAGE_SIZE) { + catalog() + .createPartitions( + identifier, + partitions.subList( + start, + Math.min(start + PARTITION_PAGE_SIZE, partitions.size()))); + } + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException( + String.format( + "Failed to register partitions for managed format table %s.", + identifier), + e); + } finally { + advanceGeneration(identifier); + cache().invalidateAll(); + } + } + + private List loadPartitions(@Nullable String partitionNamePattern) { + List partitions = new ArrayList<>(); + Set seenPageTokens = new HashSet<>(); + try { + Catalog catalog = catalog(); + String pageToken = null; + do { + PagedList page = + catalog.listPartitionsPaged( + identifier, PARTITION_PAGE_SIZE, pageToken, partitionNamePattern); + partitions.addAll(page.getElements()); + pageToken = page.getNextPageToken(); + if (StringUtils.isNotEmpty(pageToken) && !seenPageTokens.add(pageToken)) { + throw new IllegalStateException( + String.format( + "Catalog returned repeated partition page token '%s' for managed format table %s.", + pageToken, identifier.getFullName())); + } + } while (StringUtils.isNotEmpty(pageToken)); + return Collections.unmodifiableList(partitions); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException( + String.format( + "Failed to list partitions for managed format table %s.", identifier), + e); + } + } + + private synchronized Catalog catalog() { + if (catalog == null) { + catalog = catalogLoader.load(); + } + return catalog; + } + + private synchronized Cache> cache() { + if (partitionCache == null) { + partitionCache = + Caffeine.newBuilder() + .expireAfterWrite(PARTITION_CACHE_TTL) + .maximumSize(MAX_CACHED_PATTERNS) + .executor(Runnable::run) + .build(); + } + return partitionCache; + } + + private String partitionCacheKey(@Nullable String partitionNamePattern) { + // Partition mutations advance a process-local generation so providers in this JVM skip + // stale entries. Providers in other JVMs converge when PARTITION_CACHE_TTL expires. + return generation().get() + "\000" + partitionNamePattern; + } + + private AtomicLong generation() { + return GENERATIONS.get(identifier.getFullName(), ignored -> new AtomicLong()); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java index ed98c6ba50ff..8d38388907c5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java @@ -48,7 +48,6 @@ import java.util.List; import java.util.Map; import java.util.Set; -import java.util.StringJoiner; import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition; @@ -63,6 +62,10 @@ public class FormatTableCommit implements BatchTableCommit { protected boolean overwrite = false; private Catalog hiveCatalog; private Identifier tableIdentifier; + @Nullable private final FormatTableCatalogProvider catalogProvider; + // Volatile: commit() and abort() may run on different threads when callers drive the API + // directly; abort must observe that registration already began to keep its no-op guarantee. + private volatile boolean partitionRegistrationStarted; public FormatTableCommit( String location, @@ -74,6 +77,30 @@ public FormatTableCommit( @Nullable Map staticPartitions, @Nullable String syncHiveUri, CatalogContext catalogContext) { + this( + location, + partitionKeys, + fileIO, + formatTablePartitionOnlyValueInPath, + overwrite, + tableIdentifier, + staticPartitions, + syncHiveUri, + catalogContext, + null); + } + + public FormatTableCommit( + String location, + List partitionKeys, + FileIO fileIO, + boolean formatTablePartitionOnlyValueInPath, + boolean overwrite, + Identifier tableIdentifier, + @Nullable Map staticPartitions, + @Nullable String syncHiveUri, + CatalogContext catalogContext, + @Nullable FormatTableCatalogProvider catalogProvider) { this.location = location; this.fileIO = fileIO; this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath; @@ -82,6 +109,7 @@ public FormatTableCommit( this.overwrite = overwrite; this.partitionKeys = partitionKeys; this.tableIdentifier = tableIdentifier; + this.catalogProvider = catalogProvider; if (syncHiveUri != null) { try { Options options = new Options(); @@ -143,7 +171,9 @@ public void commit(List commitMessages) { for (TwoPhaseOutputStream.Committer committer : committers) { committer.commit(this.fileIO); - if (partitionKeys != null && !partitionKeys.isEmpty() && hiveCatalog != null) { + if (partitionKeys != null + && !partitionKeys.isEmpty() + && (hiveCatalog != null || catalogProvider != null)) { partitionSpecs.add( extractPartitionSpecFromPath( committer.targetPath().getParent(), partitionKeys)); @@ -152,6 +182,14 @@ public void commit(List commitMessages) { for (TwoPhaseOutputStream.Committer committer : committers) { committer.clean(this.fileIO); } + if (catalogProvider != null && !partitionSpecs.isEmpty()) { + partitionRegistrationStarted = true; + try { + catalogProvider.createPartitions(new ArrayList<>(partitionSpecs)); + } catch (RuntimeException e) { + throw partitionRegistrationFailure(e); + } + } for (Map partitionSpec : partitionSpecs) { if (hiveCatalog != null) { try { @@ -172,11 +210,28 @@ public void commit(List commitMessages) { } } catch (Exception e) { - this.abort(commitMessages); - throw new RuntimeException(e); + if (!partitionRegistrationStarted) { + this.abort(commitMessages); + throw new RuntimeException(e); + } + throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e); } } + private RuntimeException partitionRegistrationFailure(RuntimeException cause) { + String tableName = tableIdentifier.getFullName(); + return new RuntimeException( + String.format( + "Managed partition registration failed for %s after data files were " + + "committed. Committed data files were preserved because the " + + "catalog state may be ambiguous. Verify the catalog partition " + + "metadata and, if partitions are missing, re-run a partition " + + "metadata sync for %s (e.g. the Spark procedure " + + "sys.sync_format_table_metadata or MSCK REPAIR TABLE).", + tableName, tableName), + cause); + } + private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException { Method hiveCreatePartitionsInHmsMethod = hiveCatalog @@ -192,12 +247,68 @@ private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException private LinkedHashMap extractPartitionSpecFromPath( Path partitionPath, List partitionKeys) { - if (formatTablePartitionOnlyValueInPath) { - return PartitionPathUtils.extractPartitionSpecFromPathOnlyValue( - partitionPath, partitionKeys); - } else { - return PartitionPathUtils.extractPartitionSpecFromPath(partitionPath); + // The writer always lays partitions out as //.../, so the partition + // spec is exactly the trailing partitionKeys.size() components of the partition + // directory. Never walk beyond them: the table location itself may contain foreign + // 'k=v' segments that must not leak into the registered spec. Registered values are + // RAW (unescaped); readers re-escape them when probing partition directories. + String[] components = new String[partitionKeys.size()]; + Path current = partitionPath; + for (int i = partitionKeys.size() - 1; i >= 0; i--) { + if (current == null || current.getName().isEmpty()) { + throw new IllegalArgumentException( + String.format( + "Partition path '%s' has fewer than %s directory levels required " + + "by partition keys %s of table %s.", + partitionPath, + partitionKeys.size(), + partitionKeys, + tableIdentifier.getFullName())); + } + components[i] = current.getName(); + current = current.getParent(); + } + + LinkedHashMap partitionSpec = new LinkedHashMap<>(); + for (int i = 0; i < partitionKeys.size(); i++) { + String expectedKey = partitionKeys.get(i); + String component = components[i]; + if (formatTablePartitionOnlyValueInPath) { + partitionSpec.put(expectedKey, PartitionPathUtils.unescapePathName(component)); + } else { + int splitIndex = component.indexOf('='); + if (splitIndex < 0) { + throw new IllegalArgumentException( + String.format( + "Partition directory '%s' of partition path '%s' is not in " + + "'key=value' form expected for partition key '%s' " + + "of table %s.", + component, + partitionPath, + expectedKey, + tableIdentifier.getFullName())); + } + String parsedKey = + PartitionPathUtils.unescapePathName(component.substring(0, splitIndex)); + String parsedValue = + PartitionPathUtils.unescapePathName(component.substring(splitIndex + 1)); + if (!expectedKey.equals(parsedKey)) { + throw new IllegalArgumentException( + String.format( + "Partition directory '%s' of partition path '%s' declares " + + "partition key '%s' but partition key '%s' was " + + "expected at position %s for table %s.", + component, + partitionPath, + parsedKey, + expectedKey, + i, + tableIdentifier.getFullName())); + } + partitionSpec.put(expectedKey, parsedValue); + } } + return partitionSpec; } private static Path buildPartitionPath( @@ -208,24 +319,29 @@ private static Path buildPartitionPath( if (partitionSpec.isEmpty() || partitionKeys.isEmpty()) { throw new IllegalArgumentException("partitionSpec or partitionKeys is empty."); } - StringJoiner joiner = new StringJoiner("/"); + LinkedHashMap orderedSpec = new LinkedHashMap<>(); for (int i = 0; i < partitionSpec.size(); i++) { String key = partitionKeys.get(i); if (partitionSpec.containsKey(key)) { - if (formatTablePartitionOnlyValueInPath) { - joiner.add(partitionSpec.get(key)); - } else { - joiner.add(key + "=" + partitionSpec.get(key)); - } + orderedSpec.put(key, partitionSpec.get(key)); } else { throw new RuntimeException("partitionSpec does not contain key: " + key); } } - return new Path(location, joiner.toString()); + return new Path( + location, + PartitionPathUtils.generatePartitionPathUtil( + orderedSpec, formatTablePartitionOnlyValueInPath)); } @Override public void abort(List commitMessages) { + // Once partition registration has started the data files are already committed and must be + // preserved: the catalog outcome is ambiguous (see partitionRegistrationFailure), so + // discarding the files here could delete data the catalog already references. + if (partitionRegistrationStarted) { + return; + } try { for (CommitMessage commitMessage : commitMessages) { if (commitMessage instanceof TwoPhaseCommitMessage) { diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java index cb0edefc82fc..62fc6b09e3c5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java @@ -59,6 +59,7 @@ import javax.annotation.Nullable; +import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; @@ -81,9 +82,9 @@ public class FormatTableScan implements InnerTableScan { private static final Logger LOG = LoggerFactory.getLogger(FormatTableScan.class); - private final FormatTable table; - private final CoreOptions coreOptions; - @Nullable private PartitionPredicate partitionFilter; + protected final FormatTable table; + protected final CoreOptions coreOptions; + @Nullable protected PartitionPredicate partitionFilter; @Nullable private final Integer limit; private final long targetSplitSize; private final long openFileCost; @@ -142,7 +143,7 @@ public static boolean isDataFileName(String fileName) { return fileName != null && !fileName.startsWith(".") && !fileName.startsWith("_"); } - private BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { + protected BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { RowType partitionType = table.partitionType(); GenericRow row = convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName()); @@ -160,7 +161,11 @@ public List splits() { LinkedHashMap partitionSpec = pair.getKey(); BinaryRow partitionRow = toPartitionRow(partitionSpec); if (partitionFilter == null || partitionFilter.test(partitionRow)) { - splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); + try { + splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow)); + } catch (FileNotFoundException e) { + onPartitionFileNotFound(partitionSpec, pair.getValue(), e); + } } } } else { @@ -177,7 +182,7 @@ public List splits() { } } - List, Path>> findPartitions() { + protected List, Path>> findPartitions() { LOG.debug( "Find partitions for format table {}, partition filter: {}", table.name(), @@ -222,6 +227,14 @@ List, Path>> findPartitions() { } } + protected void onPartitionFileNotFound( + LinkedHashMap partitionSpec, + Path partitionPath, + FileNotFoundException exception) + throws IOException { + throw exception; + } + protected static List, Path>> generatePartitions( List partitionKeys, RowType partitionType, diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java new file mode 100644 index 000000000000..7e16dee0839e --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java @@ -0,0 +1,209 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.utils.Pair; +import org.apache.paimon.utils.PartitionPathUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +/** A format table scan whose partition visibility is owned by a catalog. */ +public class ManagedFormatTableScan extends FormatTableScan { + + private static final Logger LOG = LoggerFactory.getLogger(ManagedFormatTableScan.class); + + private final FormatTableCatalogProvider catalogProvider; + + public ManagedFormatTableScan( + FormatTable table, + @Nullable PartitionPredicate partitionFilter, + @Nullable Integer limit) { + super(table, partitionFilter, limit); + FormatTableCatalogProvider provider = table.catalogProvider(); + if (provider == null) { + throw new IllegalStateException( + String.format( + "Managed format table %s has no catalog partition provider.", + table.fullName())); + } + this.catalogProvider = provider; + } + + @Override + protected List, Path>> findPartitions() { + String partitionNamePattern = createPartitionNamePattern(); + List partitions = catalogProvider.listPartitions(partitionNamePattern); + if (partitions.isEmpty() && partitionNamePattern == null) { + warnIfFilesystemPartitionsExist(); + } + List, Path>> result = new ArrayList<>(partitions.size()); + Path tablePath = new Path(table.location()); + // Do not trust the catalog to be duplicate-free: a repeated spec would double every split + // of that partition and silently duplicate query results. + Set seenPartitionPaths = new HashSet<>(partitions.size()); + for (Partition partition : partitions) { + LinkedHashMap partitionSpec = normalizeSpec(partition.spec()); + String partitionPath = + PartitionPathUtils.generatePartitionPathUtil( + partitionSpec, coreOptions.formatTablePartitionOnlyValueInPath()); + if (!seenPartitionPaths.add(partitionPath)) { + continue; + } + result.add(Pair.of(partitionSpec, new Path(tablePath, partitionPath))); + } + return result; + } + + @Override + public List listPartitionEntries() { + List partitions = catalogProvider.listPartitions(null); + if (partitions.isEmpty()) { + warnIfFilesystemPartitionsExist(); + } + List entries = new ArrayList<>(partitions.size()); + for (Partition partition : partitions) { + entries.add( + new PartitionEntry( + toPartitionRow(normalizeSpec(partition.spec())), + partition.recordCount(), + partition.fileSizeInBytes(), + partition.fileCount(), + partition.lastFileCreationTime(), + partition.totalBuckets())); + } + return entries; + } + + /** + * Warn when the catalog knows no partitions but the table directory contains subdirectories: + * typically a table that predates managed partition support (or was written by a client that + * does not register partitions) and needs a metadata sync before its data becomes visible. + */ + private void warnIfFilesystemPartitionsExist() { + try { + for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) { + if (status.isDir() && !status.getPath().getName().startsWith(".")) { + LOG.warn( + "Managed format table {} has no partitions registered in the catalog " + + "but its location {} contains directories. Data written " + + "before enabling catalog-managed partitions (or by clients " + + "that do not register partitions) is invisible until the " + + "partition metadata is synced, e.g. with the Spark procedure " + + "sys.sync_format_table_metadata or MSCK REPAIR TABLE.", + table.fullName(), + table.location()); + return; + } + } + } catch (IOException ignored) { + // Best-effort hint only; never fail or slow down the scan because of it. + } + } + + /** + * Build the partition-name prefix pattern pushed down to the catalog, or {@code null} to list + * all partitions without pushdown. See {@link + * PartitionPathUtils#buildPartitionNamePrefixPattern} for the pattern contract; this returns + * {@code null} when there is no partition predicate, when the predicate has no leading equality + * prefix, or when the prefix cannot be expressed in the pattern contract. + */ + @Nullable + String createPartitionNamePattern() { + Optional predicate = extractPartitionPredicate(partitionFilter); + if (!predicate.isPresent()) { + return null; + } + + Map equalityPrefix = + extractLeadingEqualityPartitionSpecWhenOnlyAnd( + table.partitionKeys(), predicate.get(), table.partitionType()); + return PartitionPathUtils.buildPartitionNamePrefixPattern( + table.partitionKeys(), equalityPrefix); + } + + private LinkedHashMap normalizeSpec(Map spec) { + if (spec == null + || spec.size() != table.partitionKeys().size() + || !spec.keySet().containsAll(table.partitionKeys())) { + throw corruptPartitionSpec(spec); + } + LinkedHashMap normalized = new LinkedHashMap<>(); + boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); + for (String partitionKey : table.partitionKeys()) { + String value = spec.get(partitionKey); + // Catalog metadata is not trusted for path construction. In a value-only layout, + // reject complete path components such as '.' and '..' that would escape the table. + try { + PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath); + } catch (IllegalArgumentException e) { + throw corruptPartitionSpec(spec); + } + normalized.put(partitionKey, value); + } + return normalized; + } + + private IllegalStateException corruptPartitionSpec(@Nullable Map spec) { + return new IllegalStateException( + String.format( + "Catalog returned corrupt partition metadata %s for managed format table %s; " + + "expected exactly the partition keys %s with values usable as " + + "path components.", + spec, table.fullName(), table.partitionKeys())); + } + + @Override + protected void onPartitionFileNotFound( + LinkedHashMap partitionSpec, + Path partitionPath, + FileNotFoundException exception) { + // A registered partition without a directory reads as empty, matching Hive semantics + // (e.g. ADD PARTITION before the first INSERT). Warn so genuine drift — a directory + // removed behind the catalog's back — is still discoverable. + LOG.warn( + "Partition '{}' of managed format table {} is registered in the catalog but its " + + "directory '{}' does not exist; treating the partition as empty. If the " + + "directory was removed on purpose, drop the partition or repair the " + + "metadata, e.g. with the Spark procedure sys.sync_format_table_metadata " + + "or MSCK REPAIR TABLE.", + PartitionPathUtils.generatePartitionName(partitionSpec, false), + table.fullName(), + partitionPath); + } +} diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java index fb7baf5b06b2..b17168cb3af2 100644 --- a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java @@ -109,13 +109,105 @@ public static String generatePartitionPathUtil( suffixBuf.append(escapePathName(e.getKey())); suffixBuf.append('='); } - suffixBuf.append(escapePathName(e.getValue())); + String value = e.getValue(); + validatePartitionValueForPath(value, onlyValue); + suffixBuf.append(escapePathName(value)); i++; } suffixBuf.append(Path.SEPARATOR); return suffixBuf.toString(); } + /** + * Generate a partition path without the trailing separator, e.g. {@code dt=20250101/hr=01}. + * This is the canonical partition name used when talking to a partition-managing catalog. + */ + public static String generatePartitionName( + LinkedHashMap partitionSpec, boolean onlyValue) { + String path = generatePartitionPathUtil(partitionSpec, onlyValue); + return path.endsWith(Path.SEPARATOR) + ? path.substring(0, path.length() - Path.SEPARATOR.length()) + : path; + } + + /** + * Validate that a partition value is safe for the configured path layout. In a key-value + * layout, values such as {@code "."} are part of a component such as {@code "pt=."} and are + * safe. In a value-only layout, {@code "."} and {@code ".."} are complete path components and + * would resolve to a different directory. + */ + public static void validatePartitionValueForPath(String value, boolean onlyValueInPath) { + if (value == null + || value.isEmpty() + || (onlyValueInPath && (".".equals(value) || "..".equals(value)))) { + throw new IllegalArgumentException( + String.format( + "Partition value '%s' cannot be used as a partition path component.", + value)); + } + } + + /** Conservatively validate a value when the physical partition layout is unknown. */ + public static void validatePartitionValueForPath(String value) { + validatePartitionValueForPath(value, true); + } + + /** Validate every value of a partition spec for the configured path layout. */ + public static void validatePartitionSpecForPath( + Map partitionSpec, boolean onlyValueInPath) { + for (String value : partitionSpec.values()) { + validatePartitionValueForPath(value, onlyValueInPath); + } + } + + /** Conservatively validate a spec when the physical partition layout is unknown. */ + public static void validatePartitionSpecForPath(Map partitionSpec) { + validatePartitionSpecForPath(partitionSpec, true); + } + + /** + * Build the partition-name prefix pattern pushed down to a partition-managing catalog from the + * leading equality prefix of a partition predicate. + * + *

Pattern contract (shared by every engine talking to the catalog): partition names are the + * escaped {@code key=value} form joined by {@code '/'}; {@code '%'} is the only wildcard and + * there is no escape sequence for it ({@code '_'} stays a literal). A complete spec matches the + * exact partition name; an incomplete prefix is suffixed with {@code '%'}. + * + *

Returns {@code null} whenever pushdown must be skipped and the caller should list all + * partitions instead: the equality prefix is empty, a prefix value is blank, or the escaped + * prefix contains a literal {@code '%'} that the contract cannot express. + */ + @Nullable + public static String buildPartitionNamePrefixPattern( + List partitionKeys, Map equalityPrefix) { + if (equalityPrefix.isEmpty()) { + return null; + } + LinkedHashMap orderedPrefix = new LinkedHashMap<>(); + for (String partitionKey : partitionKeys) { + if (!equalityPrefix.containsKey(partitionKey)) { + break; + } + String value = equalityPrefix.get(partitionKey); + if (StringUtils.isNullOrWhitespaceOnly(value)) { + return null; + } + orderedPrefix.put(partitionKey, value); + } + if (orderedPrefix.isEmpty()) { + return null; + } + String escapedPrefix = generatePartitionPath(orderedPrefix); + if (escapedPrefix.indexOf('%') >= 0) { + return null; + } + if (orderedPrefix.size() == partitionKeys.size()) { + return escapedPrefix.substring(0, escapedPrefix.length() - 1); + } + return escapedPrefix + '%'; + } + public static List generatePartitionPaths( List> partitions, RowType partitionType) { return partitions.stream() @@ -265,12 +357,45 @@ public static LinkedHashMap extractPartitionSpecFromPath(Path cu return fullPartSpec; } + /** Extract exactly the trailing key-value components for the declared partition keys. */ + @Nullable + static LinkedHashMap extractPartitionSpecFromPath( + Path currPath, List partitionKeys) { + String[] values = new String[partitionKeys.size()]; + Path current = currPath; + for (int i = partitionKeys.size() - 1; i >= 0; i--) { + if (current == null) { + return null; + } + Matcher matcher = PARTITION_NAME_PATTERN.matcher(current.getName()); + if (!matcher.matches() + || !partitionKeys.get(i).equals(unescapePathName(matcher.group(1)))) { + return null; + } + values[i] = unescapePathName(matcher.group(2)); + current = current.getParent(); + } + + LinkedHashMap spec = new LinkedHashMap<>(); + for (int i = 0; i < partitionKeys.size(); i++) { + spec.put(partitionKeys.get(i), values[i]); + } + return spec; + } + public static LinkedHashMap extractPartitionSpecFromPathOnlyValue( Path currPath, List partitionKeys) { LinkedHashMap fullPartSpec = new LinkedHashMap<>(); String[] split = currPath.toString().split(Path.SEPARATOR); for (int i = 0; i < partitionKeys.size(); i++) { - fullPartSpec.put(partitionKeys.get(i), split[split.length - partitionKeys.size() + i]); + // Unescape the directory component so the extracted value is the RAW partition value, + // consistent with the key=value branch (extractPartitionSpecFromPath) and with the + // values the write path registers into a partition-managing catalog. Without this, + // directories containing escaped characters (e.g. a%3Ab) would round-trip to a + // different value than the one registered (a:b). + fullPartSpec.put( + partitionKeys.get(i), + unescapePathName(split[split.length - partitionKeys.size() + i])); } return fullPartSpec; } @@ -330,8 +455,9 @@ public static List, Path>> searchPartSpecAndP part.getPath(), partitionKeys), part.getPath())); } else { - LinkedHashMap spec = extractPartitionSpecFromPath(part.getPath()); - if (spec.size() != partitionKeys.size()) { + LinkedHashMap spec = + extractPartitionSpecFromPath(part.getPath(), partitionKeys); + if (spec == null) { // illegal path, for example: /path/to/table/tmp/unknown, path without "=" continue; } @@ -360,8 +486,17 @@ private static FileStatus[] getFileStatusRecurse( GenericRow values = partitionType == null ? null : new GenericRow(partitionType.getFieldCount()); + FileStatus fileStatus; + try { + fileStatus = fileIO.getFileStatus(path); + } catch (FileNotFoundException e) { + // A missing root simply means the table has no partitions yet. + return new FileStatus[0]; + } catch (IOException e) { + throw new RuntimeException("Failed to list files in " + path, e); + } + try { - FileStatus fileStatus = fileIO.getFileStatus(path); // Skip partition levels already fixed by the scan-path prefix. int levelOffset = partitionKeys.size() - expectLevel; listStatusRecursively( @@ -378,9 +513,10 @@ private static FileStatus[] getFileStatusRecurse( defaultPartValue, levelOffset, values); - } catch (FileNotFoundException e) { - return new FileStatus[0]; } catch (IOException e) { + // Never degrade a mid-scan failure into an empty listing: callers diff this result + // against partition metadata and an incomplete listing would deregister partitions + // that still exist. throw new RuntimeException("Failed to list files in " + path, e); } @@ -412,7 +548,15 @@ private static void listStatusRecursively( } if (fileStatus.isDir()) { - for (FileStatus stat : fileIO.listStatus(fileStatus.getPath())) { + FileStatus[] children; + try { + children = fileIO.listStatus(fileStatus.getPath()); + } catch (FileNotFoundException e) { + // The directory vanished after the parent listed it: the partitions beneath it + // are gone, skipping just this subtree keeps the rest of the listing complete. + return; + } + for (FileStatus stat : children) { int partitionKeyIndex = levelOffset + level; String partitionKey = partitionKeys.get(partitionKeyIndex); diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java index 0961050f110d..3e91251ae812 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java @@ -368,6 +368,32 @@ public void testCreatePartitionsWithIgnoreIfExistsInvalidatesPartitionCache() th assertThat(catalog.listPartitions(identifier)).containsExactly(created); } + @Test + public void testCreatePartitionsFailureInvalidatesPartitionCache() throws Exception { + Catalog wrapped = Mockito.mock(Catalog.class); + TestableCachingCatalog catalog = + new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker); + Identifier identifier = new Identifier("db", "tbl"); + Map spec = singletonMap("dt", "20260717"); + Partition created = new Partition(spec, 0, 0, 0, 0, -1, false); + List stored = new ArrayList<>(); + when(wrapped.listPartitions(identifier)).thenAnswer(ignored -> new ArrayList<>(stored)); + RuntimeException responseLost = new RuntimeException("response lost"); + Mockito.doAnswer( + ignored -> { + stored.add(created); + throw responseLost; + }) + .when(wrapped) + .createPartitions(identifier, singletonList(spec)); + + assertThat(catalog.listPartitions(identifier)).isEmpty(); + assertThatThrownBy(() -> catalog.createPartitions(identifier, singletonList(spec))) + .isSameAs(responseLost); + + assertThat(catalog.listPartitions(identifier)).containsExactly(created); + } + @Test public void testDeadlock() throws Exception { Catalog underlyCatalog = this.catalog; diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java new file mode 100644 index 000000000000..edcc79af157c --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.catalog; + +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.TableSchema; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.Table; +import org.apache.paimon.types.DataTypes; + +import org.junit.jupiter.api.Test; + +import static org.apache.paimon.CoreOptions.FILE_FORMAT; +import static org.apache.paimon.CoreOptions.FORMAT_TABLE_IMPLEMENTATION; +import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE; +import static org.apache.paimon.CoreOptions.PATH; +import static org.apache.paimon.CoreOptions.TYPE; +import static org.apache.paimon.TableType.FORMAT_TABLE; +import static org.apache.paimon.catalog.CatalogUtils.loadTable; +import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Tests for {@link CatalogUtils}. */ +class CatalogUtilsTest { + + @Test + void testRejectEngineImplementationForManagedFormatTableOnCreate() { + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .option(TYPE.key(), FORMAT_TABLE.toString()) + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(FORMAT_TABLE_IMPLEMENTATION.key(), "engine") + .build(); + + assertThatThrownBy(() -> validateCreateTable(schema, false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()) + .hasMessageContaining(FORMAT_TABLE_IMPLEMENTATION.key()); + } + + @Test + void testEngineImplementationManagedFormatTableLoadsAsUnmanaged() throws Exception { + // The option combination is invalid for managed partitions, but an existing table must + // stay loadable: it degrades to an unmanaged format table instead of failing. + Table table = loadManagedFormatTable("engine", true); + assertUnmanagedFormatTable(table); + } + + @Test + void testManagedFormatTableOutsideCapableCatalogLoadsAsUnmanaged() throws Exception { + // A catalog that cannot manage partitions (e.g. Hive) must not reject existing tables that + // carry the (previously inert) option; it loads them as unmanaged format tables. + Table table = loadManagedFormatTable("paimon", false); + assertUnmanagedFormatTable(table); + } + + @Test + void testLoadManagedFormatTableFromCapableCatalog() throws Exception { + Table table = loadManagedFormatTable("paimon", true); + assertThat(table).isInstanceOf(FormatTable.class); + FormatTable formatTable = (FormatTable) table; + assertThat( + new org.apache.paimon.CoreOptions(formatTable.options()) + .partitionedTableInMetastore()) + .isTrue(); + assertThat(formatTable.catalogProvider()).isNotNull(); + } + + private static void assertUnmanagedFormatTable(Table table) { + assertThat(table).isInstanceOf(FormatTable.class); + FormatTable formatTable = (FormatTable) table; + assertThat( + new org.apache.paimon.CoreOptions(formatTable.options()) + .partitionedTableInMetastore()) + .isFalse(); + assertThat(formatTable.catalogProvider()).isNull(); + } + + private static Table loadManagedFormatTable( + String formatTableImplementation, boolean supportsManagedPartitions) throws Exception { + Identifier identifier = Identifier.create("managed_db", "managed_table"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .option(TYPE.key(), FORMAT_TABLE.toString()) + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(FORMAT_TABLE_IMPLEMENTATION.key(), formatTableImplementation) + .option(FILE_FORMAT.key(), "parquet") + .option(PATH.key(), "file:///tmp/managed_db.db/managed_table") + .build(); + TableMetadata metadata = + new TableMetadata(TableSchema.create(0, schema), false, "managed-table-id"); + Catalog catalog = mock(Catalog.class); + when(catalog.catalogLoader()).thenReturn(() -> catalog); + when(catalog.supportsManagedFormatTablePartitions()).thenReturn(supportsManagedPartitions); + + return loadTable( + catalog, + identifier, + path -> new LocalFileIO(), + path -> new LocalFileIO(), + ignored -> metadata, + null, + null, + null, + supportsManagedPartitions); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java index 04f67c018d1b..7bb122e1fea6 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java @@ -18,9 +18,15 @@ package org.apache.paimon.catalog; +import org.apache.paimon.CoreOptions; +import org.apache.paimon.TableType; import org.apache.paimon.fs.Path; import org.apache.paimon.options.Options; import org.apache.paimon.schema.Schema; +import org.apache.paimon.schema.SchemaChange; +import org.apache.paimon.schema.SchemaManager; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.Table; import org.apache.paimon.types.DataTypes; import org.apache.paimon.shade.guava30.com.google.common.collect.Lists; @@ -28,6 +34,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; /** Tests for {@link FileSystemCatalog}. */ @@ -77,4 +84,49 @@ public void testAlterDatabase() throws Exception { false)) .isInstanceOf(UnsupportedOperationException.class); } + + @Test + public void testResetManagedPartitionsOnStrandedFormatTable() throws Exception { + String database = "stranded_managed_format_table_db"; + Identifier identifier = Identifier.create(database, "stranded_format_table"); + catalog.createDatabase(database, false); + // Write the schema file directly to simulate a format table stranded with the + // REST-only option in a filesystem catalog. + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString()) + .option(CoreOptions.FILE_FORMAT.key(), "parquet") + .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true") + .build(); + Path tablePath = + ((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog)) + .getTableLocation(identifier); + new SchemaManager(fileIO, tablePath).createTable(schema); + // A catalog that cannot manage partitions must still load such a stranded table: the + // option is ignored and the table degrades to an unmanaged format table (readable), rather + // than becoming permanently unreadable. + Table stranded = catalog.getTable(identifier); + assertThat(stranded).isInstanceOf(FormatTable.class); + assertThat(new CoreOptions(stranded.options()).partitionedTableInMetastore()).isFalse(); + + // Alters that do not touch managed-sensitive options skip the managed validation. + catalog.alterTable( + identifier, + Lists.newArrayList(SchemaChange.setOption("custom.unrelated-option", "value")), + false); + + // The documented remediation must succeed on a non-REST catalog. + catalog.alterTable( + identifier, + Lists.newArrayList( + SchemaChange.removeOption(CoreOptions.METASTORE_PARTITIONED_TABLE.key())), + false); + + assertThat(catalog.getTable(identifier).options()) + .doesNotContainKey(CoreOptions.METASTORE_PARTITIONED_TABLE.key()) + .containsEntry("custom.unrelated-option", "value"); + } } diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java index b4af3194b72f..c05f45875ed4 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java @@ -26,6 +26,8 @@ import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.catalog.Identifier; import org.apache.paimon.catalog.TableQueryAuthResult; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.options.CatalogOptions; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.FieldRef; @@ -46,6 +48,8 @@ import org.apache.paimon.rest.exceptions.NotImplementedException; import org.apache.paimon.rest.responses.ConfigResponse; import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.format.FormatTableCatalogProvider; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.JsonSerdeUtil; @@ -247,6 +251,96 @@ void testManagedFormatTablePartitionListingDoesNotFallback() throws Exception { .isInstanceOf(NotImplementedException.class); } + @Test + void testCreatePartitionFailureInvalidatesManagedListingCache() throws Exception { + Identifier identifier = createManagedFormatTable(); + FormatTable table = (FormatTable) restCatalog.getTable(identifier); + FormatTableCatalogProvider provider = table.catalogProvider(); + assertThat(provider).isNotNull(); + assertThat(provider.listPartitions(null)).isEmpty(); + Map partition = Collections.singletonMap("dt", "20260717"); + restCatalogServer.failNextCreatePartitionsResponse(); + + assertThatThrownBy( + () -> + restCatalog.createPartitions( + identifier, Collections.singletonList(partition))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Injected response failure"); + + assertThat(provider.listPartitions(null)) + .extracting(org.apache.paimon.partition.Partition::spec) + .containsExactly(partition); + } + + @Test + void testDropPartitionFailureInvalidatesManagedListingCache() throws Exception { + Identifier identifier = createManagedFormatTable(); + FormatTable table = (FormatTable) restCatalog.getTable(identifier); + FormatTableCatalogProvider provider = table.catalogProvider(); + assertThat(provider).isNotNull(); + Map partition = Collections.singletonMap("dt", "20260717"); + restCatalog.createPartitions(identifier, Collections.singletonList(partition)); + assertThat(provider.listPartitions(null)) + .extracting(org.apache.paimon.partition.Partition::spec) + .containsExactly(partition); + restCatalogServer.failNextDropPartitionsResponse(); + + assertThatThrownBy( + () -> + restCatalog.dropPartitions( + identifier, Collections.singletonList(partition))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Injected response failure"); + + assertThat(provider.listPartitions(null)).isEmpty(); + } + + @Test + void testRejectExternalManagedFormatTableBeforeCreate() throws Exception { + Identifier identifier = Identifier.create("db1", "external_managed_format_table"); + restCatalog.createDatabase(identifier.getDatabaseName(), true); + String externalPath = dataPath + "/external-managed-format-table"; + Schema schema = + Schema.newBuilder() + .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString()) + .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true") + .option(CoreOptions.FILE_FORMAT.key(), "parquet") + .option(CoreOptions.PATH.key(), externalPath) + .column("id", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .build(); + + assertThatThrownBy(() -> restCatalog.createTable(identifier, schema, false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("internal table"); + assertThat(restCatalog.listTables(identifier.getDatabaseName())) + .doesNotContain(identifier.getTableName()); + assertThat(LocalFileIO.create().exists(new Path(externalPath))).isFalse(); + } + + @Test + void testRoundTrippedManagedFormatTableReplacePassesClientValidation() throws Exception { + Identifier identifier = createManagedFormatTable(); + FormatTable existing = (FormatTable) restCatalog.getTable(identifier); + Schema replacement = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("dt", DataTypes.STRING()) + .partitionKeys("dt") + .options(existing.options()) + .build(); + assertThat(replacement.options()) + .containsEntry(CoreOptions.PATH.key(), existing.location()); + + // The mock service does not implement Format Table replacement. Reaching that response + // proves the REST client accepted the unchanged synthetic path from the loaded table. + assertThatThrownBy(() -> restCatalog.replaceTable(identifier, replacement, false)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("replaceTable does not support format tables"); + } + private Identifier createManagedFormatTable() throws Exception { Identifier identifier = Identifier.create("db1", "managed_partition_table"); restCatalog.createDatabase(identifier.getDatabaseName(), true); diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java index 3a0e1539eea1..385b27cb9fbf 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java @@ -212,6 +212,9 @@ public class RESTCatalogServer { private final List> receivedHeaders = new ArrayList<>(); private volatile boolean partitionListingSupported = true; + private volatile boolean failNextCreatePartitionsResponse; + private volatile boolean failNextDropPartitionsResponse; + private volatile int tableGetCount; public RESTCatalogServer( String dataPath, AuthProvider authProvider, ConfigResponse config, String warehouse) { @@ -279,6 +282,22 @@ public void setPartitionListingSupported(boolean partitionListingSupported) { this.partitionListingSupported = partitionListingSupported; } + public void failNextCreatePartitionsResponse() { + this.failNextCreatePartitionsResponse = true; + } + + public void failNextDropPartitionsResponse() { + this.failNextDropPartitionsResponse = true; + } + + public int tableGetCount() { + return tableGetCount; + } + + public void resetTableGetCount() { + tableGetCount = 0; + } + public void addNoPermissionDatabase(String database) { noPermissionDatabases.add(database); } @@ -1733,6 +1752,7 @@ private MockResponse tableHandle(String method, String data, Identifier identifi } switch (method) { case "GET": + tableGetCount++; TableMetadata tableMetadata; if (identifier.isSystemTable()) { TableSchema schema = catalog.loadTableSchema(identifier); @@ -1897,6 +1917,16 @@ private MockResponse partitionsApiHandle( existed.add(spec); } } + if (failNextCreatePartitionsResponse) { + failNextCreatePartitionsResponse = false; + return mockResponse( + new ErrorResponse( + ErrorResponse.RESOURCE_TYPE_PARTITION, + tableIdentifier.getFullName(), + "Injected response failure after creating partitions.", + 400), + 400); + } return mockResponse(new CreatePartitionsResponse(created, existed), 200); default: return new MockResponse().setResponseCode(404); @@ -1940,6 +1970,16 @@ private MockResponse dropPartitionsHandle(String data, Identifier tableIdentifie } return false; }); + if (failNextDropPartitionsResponse) { + failNextDropPartitionsResponse = false; + return mockResponse( + new ErrorResponse( + ErrorResponse.RESOURCE_TYPE_PARTITION, + tableIdentifier.getFullName(), + "Injected response failure after dropping partitions.", + 400), + 400); + } return mockResponse(new DropPartitionsResponse(dropped, missing), 200); } diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java index 504dc6fab6ec..3f13c1cab844 100644 --- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java @@ -76,6 +76,7 @@ import org.apache.paimon.schema.SchemaChange; import org.apache.paimon.schema.SchemaManager; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FormatTable; import org.apache.paimon.table.Instant; import org.apache.paimon.table.Table; import org.apache.paimon.table.TableSnapshot; @@ -1639,6 +1640,63 @@ void testDropPartitionsLoadsNonManagedTableOnce() throws Exception { Mockito.verify(catalogSpy, Mockito.times(1)).getTable(identifier); } + @Test + void testManagedFormatTableCommitAndScanMatchesFileSystemMode() throws Exception { + Identifier identifier = Identifier.create("format_partition_db", "managed_scan_table"); + catalog.createDatabase(identifier.getDatabaseName(), true); + catalog.createTable( + identifier, + Schema.newBuilder() + .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString()) + .option(METASTORE_PARTITIONED_TABLE.key(), "true") + .option(CoreOptions.FILE_FORMAT.key(), "csv") + .column("id", DataTypes.INT()) + .column("year", DataTypes.INT()) + .column("month", DataTypes.INT()) + .partitionKeys("year", "month") + .build(), + false); + + FormatTable managedTable = (FormatTable) catalog.getTable(identifier); + BatchWriteBuilder writeBuilder = managedTable.newBatchWriteBuilder(); + try (BatchTableWrite write = writeBuilder.newWrite(); + BatchTableCommit commit = writeBuilder.newCommit()) { + write.write(GenericRow.of(1, 2024, 10)); + write.write(GenericRow.of(2, 2025, 10)); + write.write(GenericRow.of(3, 2025, 11)); + commit.commit(write.prepareCommit()); + } + + List> registeredPartitions = + Arrays.asList( + ImmutableMap.of("year", "2024", "month", "10"), + ImmutableMap.of("year", "2025", "month", "10"), + ImmutableMap.of("year", "2025", "month", "11")); + assertThat(catalog.listPartitions(identifier).stream().map(Partition::spec)) + .containsExactlyInAnyOrderElementsOf(registeredPartitions); + assertThat(managedTable.newReadBuilder().newScan().getClass().getSimpleName()) + .isEqualTo("ManagedFormatTableScan"); + + Map fileSystemOptions = new HashMap<>(managedTable.options()); + fileSystemOptions.put(METASTORE_PARTITIONED_TABLE.key(), "false"); + FormatTable fileSystemTable = + FormatTable.builder() + .fileIO(managedTable.fileIO()) + .identifier(Identifier.create("format_partition_db", "filesystem_scan")) + .rowType(managedTable.rowType()) + .partitionKeys(managedTable.partitionKeys()) + .location(managedTable.location()) + .format(managedTable.format()) + .options(fileSystemOptions) + .catalogContext(managedTable.catalogContext()) + .build(); + + Map partitionFilter = singletonMap("year", "2025"); + assertThat(read(managedTable, null, null, partitionFilter, null)) + .containsExactlyInAnyOrderElementsOf( + read(fileSystemTable, null, null, partitionFilter, null)); + } + @Test void testListPartitions() throws Exception { innerTestListPartitions(true); diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java new file mode 100644 index 000000000000..29fe9f4fd7fe --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java @@ -0,0 +1,185 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table; + +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.table.format.FormatTableCatalogProvider; +import org.apache.paimon.table.format.FormatTableCommit; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; +import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE; +import static org.apache.paimon.CoreOptions.READ_BATCH_SIZE; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** Compatibility tests for the public {@link FormatTable} API. */ +class FormatTableCompatibilityTest { + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testCopyRejectsManagedPartitionStateChange(boolean managed) { + FormatTable table = formatTable(Boolean.toString(managed)); + + assertThatThrownBy( + () -> + table.copy( + Collections.singletonMap( + METASTORE_PARTITIONED_TABLE.key(), + Boolean.toString(!managed)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()); + } + + @Test + void testCopyAllowsSemanticallyEquivalentManagedPartitionOption() { + FormatTable table = formatTable("TRUE"); + + FormatTable copied = + table.copy(Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), "true")); + + assertThat(copied).isNotSameAs(table); + assertThat(copied.options()) + .containsEntry(METASTORE_PARTITIONED_TABLE.key(), Boolean.TRUE.toString()); + } + + @Test + void testCopyRejectsChangingAbsentManagedPartitionDefault() { + FormatTable table = formatTable(Collections.emptyMap(), null); + + assertThat(table.options()).doesNotContainKey(METASTORE_PARTITIONED_TABLE.key()); + assertThatThrownBy( + () -> + table.copy( + Collections.singletonMap( + METASTORE_PARTITIONED_TABLE.key(), "true"))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testManagedCopyRejectsPhysicalPartitionLayoutChange(boolean onlyValueInPath) { + Map options = new LinkedHashMap<>(); + options.put(METASTORE_PARTITIONED_TABLE.key(), "true"); + options.put( + FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), Boolean.toString(onlyValueInPath)); + FormatTable table = formatTable(options, null); + + assertThatThrownBy( + () -> + table.copy( + Collections.singletonMap( + FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), + Boolean.toString(!onlyValueInPath)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key()); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testCopyPreservesManagedStateAndCatalogProviderForUnrelatedOption(boolean managed) { + Identifier identifier = Identifier.create("managed_db", "managed_table"); + FormatTableCatalogProvider catalogProvider = + new FormatTableCatalogProvider(identifier, () -> null); + Map options = new LinkedHashMap<>(); + options.put(METASTORE_PARTITIONED_TABLE.key(), Boolean.toString(managed)); + FormatTable table = formatTable(options, catalogProvider); + + FormatTable copied = table.copy(Collections.singletonMap(READ_BATCH_SIZE.key(), "128")); + + assertThat(copied).isNotSameAs(table); + assertThat(copied.options()) + .containsEntry(READ_BATCH_SIZE.key(), "128") + .containsEntry(METASTORE_PARTITIONED_TABLE.key(), Boolean.toString(managed)); + assertThat( + new org.apache.paimon.CoreOptions(copied.options()) + .partitionedTableInMetastore()) + .isEqualTo(managed); + assertThat(copied.catalogProvider()).isSameAs(catalogProvider); + assertThat(table.options()) + .containsExactlyEntriesOf(options) + .doesNotContainKey(READ_BATCH_SIZE.key()); + } + + @Test + void testManagedCatalogExtensionKeepsExistingImplementationsCompatible() throws Exception { + assertThat(FormatTable.class.getMethod("catalogProvider").isDefault()).isTrue(); + assertThat( + FormatTable.FormatTableImpl.class.getConstructor( + FileIO.class, + Identifier.class, + RowType.class, + List.class, + String.class, + FormatTable.Format.class, + Map.class, + String.class, + CatalogContext.class)) + .isNotNull(); + assertThat( + FormatTableCommit.class.getConstructor( + String.class, + List.class, + FileIO.class, + boolean.class, + boolean.class, + Identifier.class, + Map.class, + String.class, + CatalogContext.class)) + .isNotNull(); + } + + private static FormatTable formatTable(String managed) { + return formatTable( + Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), managed), null); + } + + private static FormatTable formatTable( + Map options, FormatTableCatalogProvider catalogProvider) { + return FormatTable.builder() + .fileIO(LocalFileIO.create()) + .identifier(Identifier.create("managed_db", "managed_table")) + .rowType( + RowType.builder() + .field("id", DataTypes.INT()) + .field("dt", DataTypes.STRING()) + .build()) + .partitionKeys(Collections.singletonList("dt")) + .location("file:///warehouse/managed_table") + .format(FormatTable.Format.PARQUET) + .options(options) + .catalogProvider(catalogProvider) + .build(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java new file mode 100644 index 000000000000..2bc74ca99bd6 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java @@ -0,0 +1,247 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.RenamingTwoPhaseOutputStream; +import org.apache.paimon.fs.TwoPhaseOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.table.sink.CommitMessage; +import org.apache.paimon.utils.PartitionPathUtils; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.mockito.ArgumentCaptor; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.entry; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** Tests for {@link FormatTableCommit}. */ +class FormatTableCommitTest { + + @TempDir java.nio.file.Path tempDir; + + @Test + void testPartitionRegistrationFailurePreservesCommittedFiles() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path targetPath = new Path(tablePath, "year=2025/month=10/data-1.csv"); + RenamingTwoPhaseOutputStream outputStream = + new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); + outputStream.write(1); + TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit(); + FormatTableCatalogProvider catalogProvider = mock(FormatTableCatalogProvider.class); + RuntimeException registrationFailure = + new RuntimeException("Catalog partition registration unavailable"); + doThrow(registrationFailure).when(catalogProvider).createPartitions(anyList()); + + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + false, + Identifier.create("managed_db", "managed_table"), + null, + null, + null, + catalogProvider); + CommitMessage message = new TwoPhaseCommitMessage(committer); + + assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Committed data files were preserved") + .hasMessageContaining("managed_db.managed_table") + .hasMessageContaining("MSCK REPAIR TABLE") + .hasRootCauseMessage("Catalog partition registration unavailable"); + + assertThat(fileIO.exists(targetPath)).isTrue(); + verify(catalogProvider).createPartitions(anyList()); + + // Flink calls abort on the same commit object after a failed commit. + commit.abort(Collections.singletonList(message)); + assertThat(fileIO.exists(targetPath)).isTrue(); + } + + @Test + void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class); + doThrow(new IOException("data commit failed")).when(committer).commit(fileIO); + FormatTableCatalogProvider catalogProvider = mock(FormatTableCatalogProvider.class); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + false, + Identifier.create("managed_db", "managed_table"), + null, + null, + null, + catalogProvider); + CommitMessage message = new TwoPhaseCommitMessage(committer); + + assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) + .isInstanceOf(RuntimeException.class) + .hasRootCauseMessage("data commit failed"); + + verify(committer).discard(fileIO); + verify(catalogProvider, never()).createPartitions(anyList()); + } + + @Test + void testRegistersRawPartitionValuesForEscapedPath() throws Exception { + Path tablePath = new Path(tempDir.toUri()); + LinkedHashMap rawSpec = new LinkedHashMap<>(); + rawSpec.put("year", "2025"); + rawSpec.put("month", "a b:c"); + // The writer escapes partition values when building the directory layout. + String partitionDir = PartitionPathUtils.generatePartitionPathUtil(rawSpec, false); + assertThat(partitionDir).isEqualTo("year=2025/month=a b%3Ac/"); + + FormatTableCatalogProvider catalogProvider = + commitPartitionedFile(tablePath, false, partitionDir); + + // The catalog must receive RAW values; readers re-escape them when probing directories. + assertThat(registeredSpec(catalogProvider)) + .containsExactly(entry("year", "2025"), entry("month", "a b:c")); + } + + @Test + void testForeignKeyValueSegmentsInLocationDoNotLeakIntoSpec() throws Exception { + Path tablePath = new Path(new Path(tempDir.toUri()), "env=prod/warehouse/tbl"); + + FormatTableCatalogProvider catalogProvider = + commitPartitionedFile(tablePath, false, "year=2025/month=10"); + + assertThat(registeredSpec(catalogProvider)) + .containsExactly(entry("year", "2025"), entry("month", "10")); + } + + @Test + void testValueOnlyPathUnderForeignKeyValueSegmentRegistersRawValues() throws Exception { + Path tablePath = new Path(new Path(tempDir.toUri()), "env=prod/warehouse/tbl"); + LinkedHashMap rawSpec = new LinkedHashMap<>(); + rawSpec.put("year", "2025"); + rawSpec.put("month", "a:b"); + String partitionDir = PartitionPathUtils.generatePartitionPathUtil(rawSpec, true); + assertThat(partitionDir).isEqualTo("2025/a%3Ab/"); + + FormatTableCatalogProvider catalogProvider = + commitPartitionedFile(tablePath, true, partitionDir); + + assertThat(registeredSpec(catalogProvider)) + .containsExactly(entry("year", "2025"), entry("month", "a:b")); + } + + @Test + void testMismatchedPartitionKeyInPathFailsWithClearMessage() { + Path tablePath = new Path(tempDir.toUri()); + + assertThatThrownBy(() -> commitPartitionedFile(tablePath, false, "year=2025/day=10")) + .isInstanceOf(RuntimeException.class) + .hasRootCauseInstanceOf(IllegalArgumentException.class) + .rootCause() + .hasMessageContaining("declares partition key 'day'") + .hasMessageContaining("partition key 'month' was expected"); + } + + @Test + void testValueOnlyStaticPartitionCannotEscapeTableLocation() throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path parentPath = new Path(tempDir.toUri()); + Path tablePath = new Path(parentPath, "table"); + Path siblingPath = new Path(parentPath, "keep"); + fileIO.mkdirs(tablePath); + fileIO.mkdirs(siblingPath); + Map staticPartition = Collections.singletonMap("year", ".."); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Collections.singletonList("year"), + fileIO, + true, + true, + Identifier.create("managed_db", "managed_table"), + staticPartition, + null, + null, + null); + + assertThatThrownBy(() -> commit.commit(Collections.emptyList())) + .isInstanceOf(RuntimeException.class) + .hasRootCauseInstanceOf(IllegalArgumentException.class) + .hasRootCauseMessage( + "Partition value '..' cannot be used as a partition path component."); + assertThat(fileIO.exists(tablePath)).isTrue(); + assertThat(fileIO.exists(siblingPath)).isTrue(); + } + + private FormatTableCatalogProvider commitPartitionedFile( + Path tableLocation, boolean onlyValueInPath, String partitionDir) throws Exception { + LocalFileIO fileIO = LocalFileIO.create(); + Path targetPath = new Path(new Path(tableLocation, partitionDir), "data-1.csv"); + RenamingTwoPhaseOutputStream outputStream = + new RenamingTwoPhaseOutputStream(fileIO, targetPath, false); + outputStream.write(1); + TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit(); + FormatTableCatalogProvider catalogProvider = mock(FormatTableCatalogProvider.class); + FormatTableCommit commit = + new FormatTableCommit( + tableLocation.toString(), + Arrays.asList("year", "month"), + fileIO, + onlyValueInPath, + false, + Identifier.create("managed_db", "managed_table"), + null, + null, + null, + catalogProvider); + commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); + return catalogProvider; + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static Map registeredSpec(FormatTableCatalogProvider catalogProvider) { + ArgumentCaptor>> captor = + ArgumentCaptor.forClass((Class) List.class); + verify(catalogProvider).createPartitions(captor.capture()); + assertThat(captor.getValue()).hasSize(1); + return captor.getValue().get(0); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java new file mode 100644 index 000000000000..0ca741386eed --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java @@ -0,0 +1,553 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.table.format; + +import org.apache.paimon.PagedList; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.data.BinaryString; +import org.apache.paimon.fs.FileStatus; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.PositionOutputStream; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.manifest.PartitionEntry; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.table.FormatTable; +import org.apache.paimon.table.source.Split; +import org.apache.paimon.types.DataTypes; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.PartitionPathUtils; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.FileNotFoundException; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Tests for managed format table partition discovery. */ +class ManagedFormatTableScanTest { + + private static final Identifier IDENTIFIER = Identifier.create("managed_db", "managed_table"); + + @TempDir java.nio.file.Path tempDir; + + @Test + void testProviderRegistersLargeBatchesInBoundedRequests() throws Exception { + Catalog catalog = mock(Catalog.class); + FormatTableCatalogProvider provider = provider(catalog); + List> specs = new ArrayList<>(); + for (int index = 0; index < 2500; index++) { + specs.add(Collections.singletonMap("year", String.format("y%04d", index))); + } + + provider.createPartitions(specs); + + @SuppressWarnings({"unchecked", "rawtypes"}) + org.mockito.ArgumentCaptor>> batches = + (org.mockito.ArgumentCaptor) org.mockito.ArgumentCaptor.forClass(List.class); + verify(catalog, times(3)).createPartitions(eq(IDENTIFIER), batches.capture()); + assertThat(batches.getAllValues().get(0)).hasSize(1000); + assertThat(batches.getAllValues().get(1)).hasSize(1000); + assertThat(batches.getAllValues().get(2)).hasSize(500); + assertThat( + batches.getAllValues().stream() + .flatMap(List::stream) + .collect(java.util.stream.Collectors.toList())) + .containsExactlyElementsOf(specs); + } + + @Test + void testProviderPaginatesAndCachesMatchingPattern() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition october = partition("2025", "10"); + Partition november = partition("2025", "11"); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + .thenReturn(new PagedList<>(Collections.singletonList(october), "next")); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), eq("next"), eq("year=2025/%"))) + .thenReturn(new PagedList<>(Collections.singletonList(november), null)); + + FormatTableCatalogProvider provider = provider(catalog); + + assertThat(provider.listPartitions("year=2025/%")).containsExactly(october, november); + assertThat(provider.listPartitions("year=2025/%")).containsExactly(october, november); + verify(catalog, times(1)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%"); + verify(catalog, times(1)).listPartitionsPaged(IDENTIFIER, 1000, "next", "year=2025/%"); + } + + @Test + void testProviderRejectsRepeatedPageTokenCycle() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Collections.emptyList(), "a")); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), eq("a"), isNull())) + .thenReturn(new PagedList<>(Collections.emptyList(), "b")) + .thenThrow(new AssertionError("provider requested page token 'a' twice")); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), eq("b"), isNull())) + .thenReturn(new PagedList<>(Collections.emptyList(), "a")); + + assertThatThrownBy(() -> provider(catalog).listPartitions(null)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("repeated partition page token 'a'") + .hasMessageContaining(IDENTIFIER.getFullName()); + } + + @Test + void testCreatePartitionsInvalidatesCachedPattern() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition oldPartition = partition("2025", "10"); + Partition newPartition = partition("2025", "11"); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + .thenReturn( + new PagedList<>(Collections.singletonList(oldPartition), null), + new PagedList<>(Arrays.asList(oldPartition, newPartition), null)); + + FormatTableCatalogProvider provider = provider(catalog); + assertThat(provider.listPartitions("year=2025/%")).containsExactly(oldPartition); + assertThat(provider.listPartitions("year=2025/%")).containsExactly(oldPartition); + + List> created = Collections.singletonList(newPartition.spec()); + provider.createPartitions(created); + + assertThat(provider.listPartitions("year=2025/%")) + .containsExactly(oldPartition, newPartition); + verify(catalog).createPartitions(IDENTIFIER, created); + verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%"); + } + + @Test + void testCreatePartitionsInvalidatesOtherProviderInSameProcess() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition oldPartition = partition("2025", "10"); + Partition newPartition = partition("2025", "11"); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + .thenReturn( + new PagedList<>(Collections.singletonList(oldPartition), null), + new PagedList<>(Arrays.asList(oldPartition, newPartition), null)); + FormatTableCatalogProvider readerProvider = provider(catalog); + FormatTableCatalogProvider writerProvider = provider(catalog); + + assertThat(readerProvider.listPartitions("year=2025/%")).containsExactly(oldPartition); + writerProvider.createPartitions(Collections.singletonList(newPartition.spec())); + + assertThat(readerProvider.listPartitions("year=2025/%")) + .containsExactly(oldPartition, newPartition); + verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%"); + } + + @Test + void testCreatePartitionsFailureInvalidatesOtherProviderCache() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition oldPartition = partition("2025", "10"); + Partition newPartition = partition("2025", "11"); + List storedPartitions = new ArrayList<>(); + storedPartitions.add(oldPartition); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + .thenAnswer(ignored -> new PagedList<>(new ArrayList<>(storedPartitions), null)); + List> created = Collections.singletonList(newPartition.spec()); + RuntimeException responseLost = new RuntimeException("response lost"); + doAnswer( + ignored -> { + storedPartitions.add(newPartition); + throw responseLost; + }) + .when(catalog) + .createPartitions(IDENTIFIER, created); + FormatTableCatalogProvider readerProvider = provider(catalog); + FormatTableCatalogProvider writerProvider = provider(catalog); + + assertThat(readerProvider.listPartitions("year=2025/%")).containsExactly(oldPartition); + assertThatThrownBy(() -> writerProvider.createPartitions(created)).isSameAs(responseLost); + + assertThat(readerProvider.listPartitions("year=2025/%")) + .containsExactly(oldPartition, newPartition); + verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%"); + } + + @Test + void testAdvanceGenerationInvalidatesProvidersInSameProcess() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition oldPartition = partition("2025", "10"); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + .thenReturn( + new PagedList<>(Arrays.asList(oldPartition, partition("2025", "11")), null), + new PagedList<>(Collections.singletonList(oldPartition), null)); + FormatTableCatalogProvider provider = provider(catalog); + + assertThat(provider.listPartitions("year=2025/%")).hasSize(2); + // The catalog partition DDL path (e.g. RESTCatalog.dropPartitions) advances the + // generation so same-process scans read their own writes within the cache TTL. + FormatTableCatalogProvider.advanceGeneration(IDENTIFIER); + + assertThat(provider.listPartitions("year=2025/%")).containsExactly(oldPartition); + verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%"); + } + + @Test + void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%"))) + .thenReturn( + new PagedList<>( + Arrays.asList(partition("2025", "10"), partition("2025", "11")), + null)); + TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + Path octoberFile = writeDataFile(fileIO, tablePath, "year=2025/month=10"); + Path novemberFile = writeDataFile(fileIO, tablePath, "year=2025/month=11"); + Path unregisteredFile = writeDataFile(fileIO, tablePath, "year=2025/month=12"); + FormatTable table = createTable(fileIO, tablePath, provider(catalog), false); + PredicateBuilder builder = new PredicateBuilder(table.partitionType()); + Predicate predicate = + PredicateBuilder.and(builder.equal(0, 2025), builder.greaterThan(1, 10)); + PartitionPredicate filter = + PartitionPredicate.fromPredicate(table.partitionType(), predicate); + + ManagedFormatTableScan scan = new ManagedFormatTableScan(table, filter, null); + List plannedFiles = plannedFiles(scan.plan().splits()); + + assertThat(scan.createPartitionNamePattern()).isEqualTo("year=2025/%"); + assertThat(plannedFiles).containsExactly(novemberFile); + assertThat(plannedFiles).doesNotContain(octoberFile, unregisteredFile); + assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath, "year=2025/month=11")); + } + + @Test + void testListPartitionEntriesUsesCatalogVisibility() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition catalogPartition = + new Partition(partition("2025", "11").spec(), 11L, 22L, 3L, 44L, 5, false); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Collections.singletonList(catalogPartition), null)); + TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + writeDataFile(fileIO, tablePath, "year=2025/month=11"); + writeDataFile(fileIO, tablePath, "year=2025/month=12"); + FormatTable table = createTable(fileIO, tablePath, provider(catalog), false); + + List entries = + new ManagedFormatTableScan(table, null, null).listPartitionEntries(); + + assertThat(entries) + .extracting( + entry -> entry.partition().getInt(0), entry -> entry.partition().getInt(1)) + .containsExactly(org.assertj.core.groups.Tuple.tuple(2025, 11)); + assertThat(entries.get(0).recordCount()).isEqualTo(11L); + assertThat(entries.get(0).fileSizeInBytes()).isEqualTo(22L); + assertThat(entries.get(0).fileCount()).isEqualTo(3L); + assertThat(entries.get(0).lastFileCreationTime()).isEqualTo(44L); + assertThat(entries.get(0).totalBuckets()).isEqualTo(5); + assertThat(fileIO.listedPaths).isEmpty(); + assertThat(fileIO.statusListedPaths).isEmpty(); + } + + @Test + void testUnderscoreInPartitionNameRemainsLiteralPrefix() { + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTable table = + createStringPartitionTable(fileIO, tablePath, provider(mock(Catalog.class))); + Predicate predicate = + new PredicateBuilder(table.partitionType()) + .equal(0, BinaryString.fromString("a_b")); + PartitionPredicate filter = + PartitionPredicate.fromPredicate(table.partitionType(), predicate); + + ManagedFormatTableScan scan = new ManagedFormatTableScan(table, filter, null); + + assertThat(scan.createPartitionNamePattern()).isEqualTo("year=a_b/%"); + } + + @Test + void testValueOnlyPartitionPath() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn( + new PagedList<>(Collections.singletonList(partition("2025", "11")), null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path dataFile = writeDataFile(fileIO, tablePath, "2025/11"); + FormatTable table = createTable(fileIO, tablePath, provider(catalog), true); + + List plannedFiles = + plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits()); + + assertThat(plannedFiles).containsExactly(dataFile); + } + + @Test + void testDuplicateCatalogPartitionPlansSplitsOnce() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn( + new PagedList<>( + Arrays.asList(partition("2025", "11"), partition("2025", "11")), + null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path dataFile = writeDataFile(fileIO, tablePath, "year=2025/month=11"); + FormatTable table = createTable(fileIO, tablePath, provider(catalog), false); + + List plannedFiles = + plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits()); + + assertThat(plannedFiles).containsExactly(dataFile); + } + + @Test + void testWhitespacePartitionValueIsVisible() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition whitespacePartition = partition(" ", "11"); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Collections.singletonList(whitespacePartition), null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path dataFile = + writeDataFile( + fileIO, + tablePath, + PartitionPathUtils.generatePartitionPath( + new LinkedHashMap<>(whitespacePartition.spec()))); + FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog)); + + List plannedFiles = + plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits()); + + assertThat(plannedFiles).containsExactly(dataFile); + } + + @Test + void testEmptyPartitionValueReportsCorruptMetadata() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition emptyValuePartition = partition("2025", ""); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Collections.singletonList(emptyValuePartition), null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog)); + + assertThatThrownBy(() -> new ManagedFormatTableScan(table, null, null).plan().splits()) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("corrupt partition metadata") + .hasMessageContaining(IDENTIFIER.getFullName()); + } + + @Test + void testCatalogFailureDoesNotFallBackToFileSystem() throws Exception { + Catalog catalog = mock(Catalog.class); + RuntimeException catalogFailure = + new RuntimeException("Catalog partition listing unavailable"); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenThrow(catalogFailure); + TrackingLocalFileIO fileIO = new TrackingLocalFileIO(); + Path tablePath = new Path(tempDir.toUri()); + writeDataFile(fileIO, tablePath, "year=2025/month=11"); + FormatTable table = createTable(fileIO, tablePath, provider(catalog), false); + + assertThatThrownBy(() -> new ManagedFormatTableScan(table, null, null).plan().splits()) + .isSameAs(catalogFailure); + assertThat(fileIO.listedPaths).isEmpty(); + } + + @Test + @DisplayName( + "treats a registered partition with a missing directory as an empty partition " + + "(requires real object-store validation)") + void testMissingManagedPartitionReadsAsEmpty() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn( + new PagedList<>(Collections.singletonList(partition("2025", "11")), null)); + Path tablePath = new Path(tempDir.toUri()); + Path missingPath = new Path(tablePath, "year=2025/month=11"); + LocalFileIO fileIO = + new LocalFileIO() { + @Override + public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { + assertThat(path).isEqualTo(missingPath); + throw new FileNotFoundException(path.toString()); + } + }; + FormatTable table = createTable(fileIO, tablePath, provider(catalog), false); + + // A registered partition whose directory is missing (e.g. ADD PARTITION before the first + // INSERT) must not fail the whole scan; it reads as an empty partition, matching Hive. + List plannedFiles = + plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits()); + assertThat(plannedFiles).isEmpty(); + } + + @Test + void testTraversalPartitionValueReportsCorruptMetadata() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition traversalPartition = partition("2025", ".."); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Collections.singletonList(traversalPartition), null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog), true); + + assertThatThrownBy(() -> new ManagedFormatTableScan(table, null, null).plan().splits()) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("corrupt partition metadata") + .hasMessageContaining(IDENTIFIER.getFullName()); + } + + @Test + void testDotDotValueInKeyValueLayoutRemainsVisible() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition partition = partition("2025", ".."); + when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull())) + .thenReturn(new PagedList<>(Collections.singletonList(partition), null)); + LocalFileIO fileIO = LocalFileIO.create(); + Path tablePath = new Path(tempDir.toUri()); + Path dataFile = writeDataFile(fileIO, tablePath, "year=2025/month=.."); + FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog)); + + List plannedFiles = + plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits()); + + assertThat(plannedFiles).containsExactly(dataFile); + } + + private FormatTableCatalogProvider provider(Catalog catalog) { + return new FormatTableCatalogProvider(IDENTIFIER, () -> catalog); + } + + private FormatTable createTable( + LocalFileIO fileIO, + Path tablePath, + FormatTableCatalogProvider provider, + boolean valueOnlyPath) { + RowType rowType = + RowType.builder() + .field("year", DataTypes.INT()) + .field("month", DataTypes.INT()) + .field("id", DataTypes.INT()) + .build(); + return FormatTable.builder() + .fileIO(fileIO) + .identifier(IDENTIFIER) + .rowType(rowType) + .partitionKeys(Arrays.asList("year", "month")) + .location(tablePath.toString()) + .format(FormatTable.Format.CSV) + .options( + Collections.singletonMap( + FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), + Boolean.toString(valueOnlyPath))) + .catalogProvider(provider) + .build(); + } + + private FormatTable createStringPartitionTable( + LocalFileIO fileIO, Path tablePath, FormatTableCatalogProvider provider) { + return createStringPartitionTable(fileIO, tablePath, provider, false); + } + + private FormatTable createStringPartitionTable( + LocalFileIO fileIO, + Path tablePath, + FormatTableCatalogProvider provider, + boolean valueOnlyPath) { + RowType rowType = + RowType.builder() + .field("year", DataTypes.STRING()) + .field("month", DataTypes.STRING()) + .field("id", DataTypes.INT()) + .build(); + return FormatTable.builder() + .fileIO(fileIO) + .identifier(IDENTIFIER) + .rowType(rowType) + .partitionKeys(Arrays.asList("year", "month")) + .location(tablePath.toString()) + .format(FormatTable.Format.CSV) + .options( + Collections.singletonMap( + FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), + Boolean.toString(valueOnlyPath))) + .catalogProvider(provider) + .build(); + } + + private Path writeDataFile(LocalFileIO fileIO, Path tablePath, String partitionPath) + throws IOException { + Path file = new Path(new Path(tablePath, partitionPath), "data.csv"); + fileIO.mkdirs(file.getParent()); + try (PositionOutputStream out = fileIO.newOutputStream(file, false)) { + out.write(1); + } + return file; + } + + private static Partition partition(String year, String month) { + Map spec = new LinkedHashMap<>(); + spec.put("year", year); + spec.put("month", month); + return new Partition(spec, 0, 0, 0, 0, -1, false); + } + + private static List plannedFiles(List splits) { + return splits.stream() + .map(FormatDataSplit.class::cast) + .flatMap(split -> split.files().stream()) + .map(FormatDataSplit.FileMeta::filePath) + .collect(Collectors.toList()); + } + + private static class TrackingLocalFileIO extends LocalFileIO { + + private final List listedPaths = new ArrayList<>(); + private final List statusListedPaths = new ArrayList<>(); + + @Override + public FileStatus[] listStatus(Path path) throws IOException { + statusListedPaths.add(path); + return super.listStatus(path); + } + + @Override + public FileStatus[] listFiles(Path path, boolean recursive) throws IOException { + listedPaths.add(path); + return super.listFiles(path, recursive); + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java index a21b168f54ea..6b2952778003 100644 --- a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java @@ -19,17 +19,25 @@ package org.apache.paimon.utils; import org.apache.paimon.data.GenericRow; +import org.apache.paimon.fs.Path; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.util.Arrays; +import java.util.LinkedHashMap; import static org.apache.paimon.utils.PartitionPathUtils.mightMatch; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.entry; -/** Tests for {@link PartitionPathUtils#mightMatch}. */ +/** Tests for {@link PartitionPathUtils}. */ class PartitionPathUtilsTest { private final RowType partitionType = @@ -39,6 +47,46 @@ class PartitionPathUtilsTest { .build(); private final PredicateBuilder builder = new PredicateBuilder(partitionType); + @ParameterizedTest + @ValueSource(strings = {".", ".."}) + void testValueOnlyDotSegmentIsRejected(String rawValue) { + LinkedHashMap partitionSpec = new LinkedHashMap<>(); + partitionSpec.put("pt", rawValue); + + assertThatThrownBy(() -> PartitionPathUtils.generatePartitionPathUtil(partitionSpec, true)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(rawValue); + } + + @ParameterizedTest + @ValueSource(strings = {".", ".."}) + void testKeyedDotValueKeepsCompatibleSafeLayout(String rawValue) { + LinkedHashMap partitionSpec = new LinkedHashMap<>(); + partitionSpec.put("pt", rawValue); + + String partitionPath = PartitionPathUtils.generatePartitionPathUtil(partitionSpec, false); + Path resolvedPath = new Path(new Path("file:///warehouse/table"), partitionPath); + + assertThat(partitionPath).isEqualTo("pt=" + rawValue + Path.SEPARATOR); + assertThat(PartitionPathUtils.extractPartitionSpecFromPath(resolvedPath)) + .containsExactly(entry("pt", rawValue)); + } + + @Test + void testTrailingPartitionExtractionStopsAtTableBoundary() { + Path partitionPath = new Path("file:///warehouse/env=prod/table/dt=20260718/hh=10"); + + assertThat( + PartitionPathUtils.extractPartitionSpecFromPath( + partitionPath, Arrays.asList("dt", "hh"))) + .containsExactly(entry("dt", "20260718"), entry("hh", "10")); + assertThat( + PartitionPathUtils.extractPartitionSpecFromPath( + new Path("file:///warehouse/dt=parent/table/wrong=20260718/hh=10"), + Arrays.asList("dt", "hh"))) + .isNull(); + } + @Test void testNullPredicate() { assertThat(mightMatch(null, 0, 0, GenericRow.of(2024, 5))).isTrue(); diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala index 853c7f41467b..bec591bbd7d1 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala @@ -49,6 +49,8 @@ abstract class FormatTableBatchWriteBase( extends Logging with Serializable { + @volatile protected var commitStarted: Boolean = false + protected val batchWriteBuilder: BatchWriteBuilder = { val builder = table.newBatchWriteBuilder() // todo: add test for static overwrite the whole table @@ -65,6 +67,7 @@ abstract class FormatTableBatchWriteBase( } protected def commitMessages(messages: Array[WriterCommitMessage]): Unit = { + commitStarted = true logInfo(s"Committing to FormatTable ${table.name()}") val batchTableCommit = batchWriteBuilder.newCommit() val commitMessages = WriteTaskResult.merge(messages).asJava @@ -80,6 +83,12 @@ abstract class FormatTableBatchWriteBase( } protected def abortMessages(messages: Array[WriterCommitMessage]): Unit = { + if (commitStarted) { + logWarning( + s"Skip abort cleanup for FormatTable ${table.name()} because commit has already started") + return + } + logInfo(s"Aborting write to FormatTable ${table.name()}") val batchTableCommit = batchWriteBuilder.newCommit() val commitMessages = WriteTaskResult.merge(messages).asJava diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala new file mode 100644 index 000000000000..ce4d1951e1b4 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.format + +import org.apache.paimon.spark.write.FormatTableWriteTaskResult +import org.apache.paimon.table.FormatTable +import org.apache.paimon.table.sink.{BatchTableCommit, BatchWriteBuilder} + +import org.apache.spark.SparkFunSuite +import org.apache.spark.sql.connector.write.WriterCommitMessage +import org.apache.spark.sql.types.StructType + +import java.lang.reflect.{InvocationHandler, Method, Proxy} +import java.util.concurrent.atomic.AtomicInteger + +class FormatTableBatchWriteTest extends SparkFunSuite { + + test("abort after commit starts does not clean up potentially committed files") { + val commitCalls = new AtomicInteger + val abortCalls = new AtomicInteger + val batchWrite = + new FormatTableBatchWrite(formatTable(commitCalls, abortCalls), None, None, StructType(Nil)) + val messages: Array[WriterCommitMessage] = Array(FormatTableWriteTaskResult(Seq.empty)) + + val error = intercept[RuntimeException] { + batchWrite.commit(messages) + } + assert(error.getMessage == "partition registration response lost") + + // Spark invokes abort after commit throws. The commit outcome may be ambiguous, so cleanup + // could delete files which were already committed and registered remotely. + batchWrite.abort(messages) + + assert(commitCalls.get == 1) + assert(abortCalls.get == 0) + } + + test("abort before commit delegates cleanup") { + val commitCalls = new AtomicInteger + val abortCalls = new AtomicInteger + val batchWrite = + new FormatTableBatchWrite(formatTable(commitCalls, abortCalls), None, None, StructType(Nil)) + + batchWrite.abort(Array(FormatTableWriteTaskResult(Seq.empty))) + + assert(commitCalls.get == 0) + assert(abortCalls.get == 1) + } + + private def formatTable(commitCalls: AtomicInteger, abortCalls: AtomicInteger): FormatTable = { + val tableCommit = proxy(classOf[BatchTableCommit]) { + case "commit" => + commitCalls.incrementAndGet() + throw new RuntimeException("partition registration response lost") + case "abort" => + abortCalls.incrementAndGet() + null + } + val writeBuilder = proxy(classOf[BatchWriteBuilder]) { case "newCommit" => tableCommit } + proxy(classOf[FormatTable]) { + case "newBatchWriteBuilder" => writeBuilder + case "name" => "test_db.format_table" + } + } + + private def proxy[T](clazz: Class[T])(responses: PartialFunction[String, AnyRef]): T = { + Proxy + .newProxyInstance( + clazz.getClassLoader, + Array(clazz), + new InvocationHandler { + override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = { + responses.applyOrElse( + method.getName, + (name: String) => + throw new UnsupportedOperationException(s"Unexpected $name invocation")) + } + } + ) + .asInstanceOf[T] + } +}