From bc3cae9320a7f9ae0bb3cbdde7d4e40e5cb132df Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 21 Jul 2026 01:02:25 +0800 Subject: [PATCH] [core] Let a REST catalog own the partitions of a Format Table MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A partitioned Format Table finds its partitions by walking the table directory. That is how Hive works and it stays the default, but it means a directory appearing under the table is a partition, whether or not anyone meant it to be, and that listing a large table costs a full traversal of object storage. With metastore.partitioned-table=true on an internal Format Table in a REST catalog, the catalog becomes the source of truth instead: a scan reads the partitions it has registered, and a batch write registers the ones it wrote. A directory nobody registered is not part of the table, and an empty catalog means an empty table — reads never quietly fall back to the filesystem, since that would defeat the point. A registered partition whose directory is missing reads as empty, matching Hive, where ADD PARTITION before the first insert is ordinary. The option already gates the catalog partition API for Paimon tables in a REST catalog, so a Format Table using it for the same purpose is the same idea, not a second one. On a Format Table it only takes effect in a REST catalog and only for an internal table under the paimon implementation; asking for it elsewhere fails rather than handing back a table reading its partitions from somewhere it did not ask for. Whether a table's partitions come from the catalog is then answered in one place, by whether loading it produced a partition manager. The partition metadata itself is reached through a manager the table carries: paging, per-request batching and pattern encoding are its business, and it creates one catalog per operation and closes it, so the table stays serializable without owning a client. Engine support for the partition DDL this makes possible follows separately. --- docs/docs/flink/sql-ddl.md | 23 + docs/generated/core_configuration.html | 3 +- docs/static/rest-catalog-open-api.yaml | 2 +- .../java/org/apache/paimon/CoreOptions.java | 6 +- .../apache/paimon/catalog/CatalogUtils.java | 56 ++- .../org/apache/paimon/rest/RESTCatalog.java | 39 +- .../org/apache/paimon/table/FormatTable.java | 68 ++- .../CatalogFormatTablePartitionManager.java | 200 ++++++++ .../table/format/FormatBatchWriteBuilder.java | 3 +- .../table/format/FormatTableCommit.java | 51 +- .../format/FormatTablePartitionManager.java | 69 +++ .../paimon/table/format/FormatTableScan.java | 162 ++++++- .../paimon/utils/PartitionPathUtils.java | 168 ++++++- .../paimon/catalog/CatalogUtilsTest.java | 135 ++++++ .../paimon/catalog/FileSystemCatalogTest.java | 32 ++ .../paimon/rest/MockRESTCatalogTest.java | 101 +++- .../apache/paimon/rest/RESTCatalogTest.java | 74 ++- .../table/FormatTableCompatibilityTest.java | 174 +++++++ ...atalogFormatTablePartitionManagerTest.java | 417 ++++++++++++++++ .../CatalogManagedPartitionScanTest.java | 451 ++++++++++++++++++ .../table/format/FormatTableCommitTest.java | 247 ++++++++++ .../paimon/utils/PartitionPathUtilsTest.java | 50 +- 22 files changed, 2469 insertions(+), 62 deletions(-) create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java create mode 100644 paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.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/CatalogFormatTablePartitionManagerTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java create mode 100644 paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java diff --git a/docs/docs/flink/sql-ddl.md b/docs/docs/flink/sql-ddl.md index 4964a0cc18d2..654e477079ef 100644 --- a/docs/docs/flink/sql-ddl.md +++ b/docs/docs/flink/sql-ddl.md @@ -34,6 +34,29 @@ Paimon catalogs currently support three types of metastores: See [CatalogOptions](../maintenance/configurations#catalogoptions) for detailed options when creating a catalog. +:::info + +For an internal Format Table in a REST catalog, `metastore.partitioned-table = true` makes the +catalog the source of truth for 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`. On a Paimon table +the same option keeps its existing meaning (synchronize partitions into the metastore); on a +Format Table it only takes effect in a REST catalog, and elsewhere partitions are still discovered +from the filesystem. + +**A Flink job reads only the partitions the catalog knows.** Directories written before the option +was enabled, by an older writer, or by anything that does not register what it wrote are invisible, +and a table whose catalog holds no partitions reads as empty. Flink has no SQL command to register +them: use Spark's `MSCK REPAIR TABLE` or the catalog's partition API. Flink writes on a current +version do register the partitions they produce. + +In a REST catalog, asking for catalog-managed partitions on a table that cannot have them — an +external table, or `format-table.implementation = engine` — fails. In any other catalog the option +keeps the meaning it has always had on a Format Table — none — and partitions come from the +filesystem. Removing the option needs a catalog that can alter the table; Flink's Format Table path +cannot, so use another engine. + +::: + ### 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 8875140929d6..bd362a033509 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -1041,7 +1041,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 an internal format table in a REST catalog, it also makes the catalog own the table's partitions: a scan reads the partitions registered there and a write registers the ones it wrote, instead of listing the table directory.
metastore.tag-to-partition
diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml index 2754e720d147..3956a9280649 100644 --- a/docs/static/rest-catalog-open-api.yaml +++ b/docs/static/rest-catalog-open-api.yaml @@ -1030,7 +1030,7 @@ paths: tags: - partition summary: Drop partitions - description: Unregisters partitions from the catalog. The server never deletes data files; managed format table data deletion is performed by the client afterwards. + description: Unregisters partitions from the catalog. The server never deletes data files; for a format table with catalog-managed partitions, data deletion is performed by the client afterwards. operationId: dropPartitions parameters: - name: prefix 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 0d2a5926da53..1419b7c2b26c 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -1850,7 +1850,11 @@ 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 an internal format table in a REST catalog, it also makes the " + + "catalog own the table's partitions: a scan reads the partitions " + + "registered there and a write registers the ones it wrote, instead of " + + "listing the table directory."); 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/CatalogUtils.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogUtils.java index 0c34fc39b1d8..ee793980d392 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.FormatTablePartitionManager; import org.apache.paimon.table.iceberg.IcebergTable; import org.apache.paimon.table.lance.LanceTable; import org.apache.paimon.table.object.ObjectTable; @@ -213,6 +214,41 @@ private static void validateFormatTableOptions(Options options, boolean dataToke } } + /** Validate options which are specific to a Format Table with catalog-managed partitions. */ + public static void validateCatalogManagedPartitionOptions(Map tableOptions) { + validateCatalogManagedPartitionOptions(Options.fromMap(tableOptions)); + } + + private static void validateCatalogManagedPartitionOptions(Options options) { + // The caller has already established that this is a format table with catalog-managed + // partitions; only the engine implementation is incompatible with them. + checkArgument( + options.get(FORMAT_TABLE_IMPLEMENTATION) + != CoreOptions.FormatTableImplementation.ENGINE, + "Cannot combine catalog-managed partitions (%s=true) with %s=engine: the engine " + + "implementation reads the table directory itself.", + CoreOptions.METASTORE_PARTITIONED_TABLE.key(), + FORMAT_TABLE_IMPLEMENTATION.key()); + } + + /** + * Validate a create or replace request that asks for catalog-managed partitions on a Format + * Table: the option combination must be valid and the table must be internal. Only the REST + * catalog calls this; other catalogs keep treating the option as inert. + */ + public static void validateCatalogManagedFormatTablePartitions( + Identifier identifier, Map tableOptions, boolean isExternal) { + CoreOptions options = CoreOptions.fromMap(tableOptions); + if (options.type() != TableType.FORMAT_TABLE || !options.partitionedTableInMetastore()) { + return; + } + validateCatalogManagedPartitionOptions(Options.fromMap(tableOptions)); + checkArgument( + !isExternal, + "Catalog-managed partitions are only supported for internal tables, but format table %s is external.", + identifier.getFullName()); + } + public static void validateNamePattern(Catalog catalog, String namePattern) { if (Objects.nonNull(namePattern) && !catalog.supportsListByPattern()) { throw new UnsupportedOperationException( @@ -304,7 +340,21 @@ public static Table loadTable( Function dataFileIO = metadata.isExternal() ? externalFileIO : internalFileIO; if (options.type() == TableType.FORMAT_TABLE) { - return toFormatTable(identifier, schema, dataFileIO, catalogContext); + FormatTablePartitionManager partitionManager = null; + if (options.partitionedTableInMetastore()) { + checkArgument( + isRestCatalog, + "Format table %s asks for catalog-managed partitions with %s=true, which " + + "is only available in a REST catalog.", + identifier.getFullName(), + CoreOptions.METASTORE_PARTITIONED_TABLE.key()); + validateCatalogManagedFormatTablePartitions( + identifier, schema.options(), metadata.isExternal()); + partitionManager = + FormatTablePartitionManager.create( + identifier, schema.partitionKeys(), catalog.catalogLoader()); + } + return toFormatTable(identifier, schema, dataFileIO, catalogContext, partitionManager); } if (options.type() == TableType.OBJECT_TABLE) { @@ -453,7 +503,8 @@ private static FormatTable toFormatTable( Identifier identifier, TableSchema schema, Function fileIO, - CatalogContext catalogContext) { + CatalogContext catalogContext, + @Nullable FormatTablePartitionManager partitionManager) { Map options = schema.options(); FormatTable.Format format = FormatTable.parseFormat( @@ -471,6 +522,7 @@ private static FormatTable toFormatTable( .options(options) .comment(schema.comment()) .catalogContext(catalogContext) + .partitionManager(partitionManager) .build(); } 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..efef63fc9275 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 @@ -94,6 +94,8 @@ import static org.apache.paimon.catalog.CatalogUtils.checkNotSystemTable; import static org.apache.paimon.catalog.CatalogUtils.isSystemDatabase; import static org.apache.paimon.catalog.CatalogUtils.listPartitionsFromFileSystem; +import static org.apache.paimon.catalog.CatalogUtils.validateCatalogManagedFormatTablePartitions; +import static org.apache.paimon.catalog.CatalogUtils.validateCatalogManagedPartitionOptions; import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable; import static org.apache.paimon.options.CatalogOptions.CASE_SENSITIVE; @@ -567,8 +569,12 @@ 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 catalog-managed partition combination, so validate + // the effective options rather than only the explicit ones. + validateCatalogManagedFormatTablePartitions( + identifier, schema.options(), schema.options().containsKey(PATH.key())); + createExternalTablePathIfNotExist(schema); Schema newSchema = inferSchemaIfExternalPaimonTable(schema); api.createTable(identifier, newSchema); } catch (AlreadyExistsException e) { @@ -645,8 +651,13 @@ 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 catalog-managed partition combination, so validate the + // effective options rather than only the explicit ones. Externality is not validated + // 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. + validateCatalogManagedPartitionOptions(newSchema.options()); try { - tableDefaultOptions.forEach(newSchema.options()::putIfAbsent); api.replaceTable(identifier, newSchema); } catch (NoSuchResourceException e) { if (!ignoreIfNotExists) { @@ -758,9 +769,9 @@ public void createPartitions( public void dropPartitions(Identifier identifier, List> partitions) throws TableNotExistException { Table table = getTable(identifier); - if (isCatalogManagedFormatTable(table)) { - // Managed format table: metadata-only unregistration on the server. Data deletion is - // the caller's responsibility (performed with the table FileIO afterwards). + if (hasCatalogManagedPartitions(table)) { + // Unregistering is metadata-only on the server; deleting the data is the caller's + // job, done with the table FileIO afterwards. try { api.dropPartitions(identifier, partitions, true); } catch (NoSuchResourceException e) { @@ -772,7 +783,7 @@ public void dropPartitions(Identifier identifier, List> part } return; } - // Non-managed tables keep the default truncate-based data semantics. + // Every other table keeps the default truncate-based data semantics. try (BatchTableCommit commit = table.newBatchWriteBuilder().newCommit()) { commit.truncatePartitions(partitions); } catch (Exception e) { @@ -790,7 +801,7 @@ public List listPartitions(Identifier identifier) throws TableNotExis throw new TableNoPermissionException(identifier, e); } catch (NotImplementedException e) { Table table = getTable(identifier); - if (isCatalogManagedFormatTable(table)) { + if (hasCatalogManagedPartitions(table)) { throw e; } return listPartitionsFromFileSystem(table); @@ -812,7 +823,7 @@ public PagedList listPartitionsPaged( throw new TableNoPermissionException(identifier, e); } catch (NotImplementedException e) { Table table = getTable(identifier); - if (isCatalogManagedFormatTable(table)) { + if (hasCatalogManagedPartitions(table)) { throw e; } return new PagedList<>(listPartitionsFromFileSystem(table), null); @@ -831,7 +842,7 @@ public List listPartitionsByNames( throw new TableNoPermissionException(identifier, e); } catch (NotImplementedException e) { Table table = getTable(identifier); - if (isCatalogManagedFormatTable(table)) { + if (hasCatalogManagedPartitions(table)) { throw e; } return listPartitionsFromFileSystem(table, partitions); @@ -1323,9 +1334,13 @@ RESTApi api() { return api; } - private static boolean isCatalogManagedFormatTable(Table table) { - return table instanceof FormatTable - && new CoreOptions(table.options()).partitionedTableInMetastore(); + /** + * Whether the catalog owns this table's partitions, which is exactly whether loading it + * produced a partition manager. Reading the option instead would be a second answer to the same + * question, and the two can disagree on a table built outside the catalog. + */ + private static boolean hasCatalogManagedPartitions(Table table) { + return table instanceof FormatTable && ((FormatTable) table).partitionManager() != null; } private FileIO fileIOForData(Path path, Identifier identifier) { 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..acb487e2573b 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,12 @@ package org.apache.paimon.table; +import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; +import org.apache.paimon.annotation.Experimental; 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 +32,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.FormatTablePartitionManager; import org.apache.paimon.table.sink.BatchWriteBuilder; import org.apache.paimon.table.sink.StreamWriteBuilder; import org.apache.paimon.table.source.BatchVectorSearchBuilder; @@ -56,8 +60,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. Format Tables with catalog-managed partitions use + * catalog metadata for partition visibility while retaining the same physical layout. * * @since 0.9.0 */ @@ -75,6 +80,17 @@ public interface FormatTable extends Table { CatalogContext catalogContext(); + /** + * The catalog partition registrations of this table, or null when its partitions are discovered + * from the filesystem. When non-null, catalog registration decides partition visibility: a + * partition directory that is not registered is not visible to scans. + */ + @Experimental + @Nullable + default FormatTablePartitionManager partitionManager() { + return null; + } + /** Currently supported formats. */ enum Format { ORC, @@ -115,6 +131,7 @@ class Builder { private Map options; @Nullable private String comment; private CatalogContext catalogContext; + @Nullable private FormatTablePartitionManager partitionManager; public Builder fileIO(FileIO fileIO) { this.fileIO = fileIO; @@ -161,6 +178,12 @@ public Builder catalogContext(CatalogContext catalogContext) { return this; } + @Experimental + public Builder partitionManager(@Nullable FormatTablePartitionManager partitionManager) { + this.partitionManager = partitionManager; + return this; + } + public FormatTable build() { return new FormatTableImpl( fileIO, @@ -171,7 +194,8 @@ public FormatTable build() { format, options, comment, - catalogContext); + catalogContext, + partitionManager); } } @@ -189,6 +213,7 @@ class FormatTableImpl implements FormatTable { private final Map options; @Nullable private final String comment; private CatalogContext catalogContext; + @Nullable private final FormatTablePartitionManager partitionManager; public FormatTableImpl( FileIO fileIO, @@ -199,7 +224,8 @@ public FormatTableImpl( Format format, Map options, @Nullable String comment, - CatalogContext catalogContext) { + CatalogContext catalogContext, + @Nullable FormatTablePartitionManager partitionManager) { this.fileIO = fileIO; this.identifier = identifier; this.rowType = rowType; @@ -209,6 +235,7 @@ public FormatTableImpl( this.options = options; this.comment = comment; this.catalogContext = catalogContext; + this.partitionManager = partitionManager; } @Override @@ -265,6 +292,30 @@ 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 hasCatalogManagedPartitions = partitionManager != null; + boolean persistedPartitionsFromCatalog = coreOptions.partitionedTableInMetastore(); + boolean dynamicPartitionsFromCatalog = copiedCoreOptions.partitionedTableInMetastore(); + if (persistedPartitionsFromCatalog != dynamicPartitionsFromCatalog) { + throw new IllegalArgumentException( + String.format( + "Dynamic option '%s' cannot change where a Format Table's partitions come from.", + CoreOptions.METASTORE_PARTITIONED_TABLE.key())); + } + if (hasCatalogManagedPartitions + && coreOptions.formatTablePartitionOnlyValueInPath() + != copiedCoreOptions.formatTablePartitionOnlyValueInPath()) { + throw new IllegalArgumentException( + String.format( + "Dynamic option '%s' cannot change the physical partition layout of a Format Table with catalog-managed partitions.", + CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key())); + } + if (hasCatalogManagedPartitions) { + CatalogUtils.validateCatalogManagedPartitionOptions(newOptions); + } + return new FormatTableImpl( fileIO, identifier, @@ -274,7 +325,8 @@ public FormatTable copy(Map dynamicOptions) { format, newOptions, comment, - catalogContext); + catalogContext, + partitionManager); } @Override @@ -302,6 +354,12 @@ public FullTextSearchBuilder newFullTextSearchBuilder() { public CatalogContext catalogContext() { return this.catalogContext; } + + @Override + @Nullable + public FormatTablePartitionManager partitionManager() { + return partitionManager; + } } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java new file mode 100644 index 000000000000..7085135f17ff --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManager.java @@ -0,0 +1,200 @@ +/* + * 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.CatalogLoader; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.utils.FunctionWithException; +import org.apache.paimon.utils.PartitionPathUtils; +import org.apache.paimon.utils.StringUtils; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import static org.apache.paimon.utils.Preconditions.checkArgument; + +/** + * A {@link FormatTablePartitionManager} reading and writing partitions through a {@link Catalog}. + * One catalog is created per operation and closed when it ends, so the table stays serializable and + * no client outlives the call that needed it. + */ +class CatalogFormatTablePartitionManager implements FormatTablePartitionManager { + + private static final long serialVersionUID = 1L; + + /** Bounds one request: catalog services cap the partitions a single call may carry. */ + private static final int REQUEST_SIZE = 1000; + + private final Identifier identifier; + private final List partitionKeys; + private final CatalogLoader catalogLoader; + + CatalogFormatTablePartitionManager( + Identifier identifier, List partitionKeys, CatalogLoader catalogLoader) { + this.identifier = identifier; + // Copied: the caller's list must not be able to change what counts as a leading prefix. + this.partitionKeys = Collections.unmodifiableList(new ArrayList<>(partitionKeys)); + this.catalogLoader = catalogLoader; + } + + @Override + public List listPartitions(Map prefix) { + LinkedHashMap ordered = orderedPrefix(partitionKeys, prefix); + checkArgument( + ordered.size() == prefix.size(), + "Partition prefix %s is not a leading prefix of partition keys %s of table %s.", + prefix, + partitionKeys, + identifier.getFullName()); + String pattern = PartitionPathUtils.buildPartitionNamePrefixPattern(partitionKeys, ordered); + return execute( + catalog -> { + List partitions = new ArrayList<>(); + Set seenPageTokens = new HashSet<>(); + String pageToken = null; + do { + PagedList page = + catalog.listPartitionsPaged( + identifier, REQUEST_SIZE, pageToken, pattern); + if (page.getElements() != null) { + 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 format table %s.", + pageToken, identifier.getFullName())); + } + } while (StringUtils.isNotEmpty(pageToken)); + // A catalog may ignore the pattern, so the prefix is enforced here as well. + partitions.removeIf(partition -> !matchesPrefix(partition, prefix)); + return Collections.unmodifiableList(partitions); + }, + "list partitions"); + } + + @Override + public List listPartitionsByNames(List> partitions) { + if (partitions.isEmpty()) { + return Collections.emptyList(); + } + return execute( + catalog -> { + List found = new ArrayList<>(); + for (List> batch : batches(partitions)) { + found.addAll(catalog.listPartitionsByNames(identifier, batch)); + } + return Collections.unmodifiableList(found); + }, + "list partitions by names"); + } + + @Override + public void createPartitions(List> partitions, boolean ignoreIfExists) { + if (partitions.isEmpty()) { + return; + } + execute( + catalog -> { + if (!ignoreIfExists) { + // Rejecting the whole batch when any partition exists is only meaningful + // if the batch stays one request, so a strict create is never split. + catalog.createPartitions(identifier, partitions, false); + return null; + } + // An idempotent create is safe to split: a rerun converges from a partially + // applied batch. + for (List> batch : batches(partitions)) { + catalog.createPartitions(identifier, batch, true); + } + return null; + }, + "create partitions"); + } + + @Override + public void dropPartitions(List> partitions) { + if (partitions.isEmpty()) { + return; + } + execute( + catalog -> { + // Dropping ignores missing partitions, so a partially applied batch converges + // on a rerun and may be split. + for (List> batch : batches(partitions)) { + catalog.dropPartitions(identifier, batch); + } + return null; + }, + "drop partitions"); + } + + private static boolean matchesPrefix(Partition partition, Map prefix) { + for (Map.Entry entry : prefix.entrySet()) { + if (!entry.getValue().equals(partition.spec().get(entry.getKey()))) { + return false; + } + } + return true; + } + + private static List>> batches(List> partitions) { + List>> batches = new ArrayList<>(); + for (int start = 0; start < partitions.size(); start += REQUEST_SIZE) { + batches.add( + partitions.subList(start, Math.min(start + REQUEST_SIZE, partitions.size()))); + } + return batches; + } + + private T execute(FunctionWithException operation, String what) { + try (Catalog catalog = catalogLoader.load()) { + return operation.apply(catalog); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException( + String.format( + "Failed to %s of format table %s.", what, identifier.getFullName()), + e); + } + } + + /** Order a prefix by the table partition keys, stopping at the first key it does not cover. */ + private static LinkedHashMap orderedPrefix( + List partitionKeys, Map prefix) { + LinkedHashMap ordered = new LinkedHashMap<>(); + for (String partitionKey : partitionKeys) { + if (!prefix.containsKey(partitionKey)) { + break; + } + ordered.put(partitionKey, prefix.get(partitionKey)); + } + return ordered; + } +} 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..0b510c594051 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.partitionManager()); } @Override 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..c0f356fe226d 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,7 @@ public class FormatTableCommit implements BatchTableCommit { protected boolean overwrite = false; private Catalog hiveCatalog; private Identifier tableIdentifier; + @Nullable private final FormatTablePartitionManager partitionManager; public FormatTableCommit( String location, @@ -73,7 +73,8 @@ public FormatTableCommit( Identifier tableIdentifier, @Nullable Map staticPartitions, @Nullable String syncHiveUri, - CatalogContext catalogContext) { + CatalogContext catalogContext, + @Nullable FormatTablePartitionManager partitionManager) { this.location = location; this.fileIO = fileIO; this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath; @@ -82,6 +83,7 @@ public FormatTableCommit( this.overwrite = overwrite; this.partitionKeys = partitionKeys; this.tableIdentifier = tableIdentifier; + this.partitionManager = partitionManager; if (syncHiveUri != null) { try { Options options = new Options(); @@ -143,7 +145,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 || partitionManager != null)) { partitionSpecs.add( extractPartitionSpecFromPath( committer.targetPath().getParent(), partitionKeys)); @@ -152,6 +156,11 @@ public void commit(List commitMessages) { for (TwoPhaseOutputStream.Committer committer : committers) { committer.clean(this.fileIO); } + if (partitionManager != null && !partitionSpecs.isEmpty()) { + // Concurrent writers may touch the same partition, so registration is an + // idempotent ADD rather than a strict create. + partitionManager.createPartitions(new ArrayList<>(partitionSpecs), true); + } for (Map partitionSpec : partitionSpecs) { if (hiveCatalog != null) { try { @@ -192,12 +201,25 @@ private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException private LinkedHashMap extractPartitionSpecFromPath( Path partitionPath, List partitionKeys) { - if (formatTablePartitionOnlyValueInPath) { - return PartitionPathUtils.extractPartitionSpecFromPathOnlyValue( - partitionPath, partitionKeys); - } else { - return PartitionPathUtils.extractPartitionSpecFromPath(partitionPath); + // Only the trailing partitionKeys.size() components are the spec: the table location + // itself may contain foreign 'k=v' segments that must not leak into what is registered. + LinkedHashMap partitionSpec = + formatTablePartitionOnlyValueInPath + ? PartitionPathUtils.extractPartitionSpecFromPathOnlyValue( + partitionPath, partitionKeys) + : PartitionPathUtils.extractPartitionSpecFromPath( + partitionPath, partitionKeys); + if (partitionSpec == null) { + throw new IllegalArgumentException( + String.format( + "Partition path '%s' does not end in the %s partition directories of " + + "table %s declared by partition keys %s.", + partitionPath, + partitionKeys.size(), + tableIdentifier.getFullName(), + partitionKeys)); } + return partitionSpec; } private static Path buildPartitionPath( @@ -208,20 +230,19 @@ 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 diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java new file mode 100644 index 000000000000..64b643fba26e --- /dev/null +++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionManager.java @@ -0,0 +1,69 @@ +/* + * 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.annotation.Experimental; +import org.apache.paimon.catalog.CatalogLoader; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.partition.Partition; + +import java.io.Serializable; +import java.util.List; +import java.util.Map; + +/** + * The catalog partition registrations of a single Format Table with catalog-managed partitions. + * + *

This manages registration metadata only: it never creates, deletes or resolves a partition + * directory. Partition values are raw values, never escaped directory names, and building + * directories from them — including keeping them inside the table — is the caller's job. + * + *

Paging, page tokens, name patterns and per-request batching are implementation details; every + * method here takes or returns its complete argument. + * + * @since 1.5 + */ +@Experimental +public interface FormatTablePartitionManager extends Serializable { + + /** + * List every partition whose leading partition values match {@code prefix}, or all partitions + * when it is empty. The prefix must be a contiguous leading subset of the table's partition + * keys. + */ + List listPartitions(Map prefix); + + /** Return those of the given complete partition specs that are registered. */ + List listPartitionsByNames(List> partitions); + + /** + * Register partitions. With {@code ignoreIfExists=false} the whole batch is rejected when any + * partition already exists, so such a request is never split. + */ + void createPartitions(List> partitions, boolean ignoreIfExists); + + /** Unregister partitions. Metadata only; missing partitions are ignored. */ + void dropPartitions(List> partitions); + + /** Create a manager reading and writing partitions through a {@link CatalogLoader}. */ + static FormatTablePartitionManager create( + Identifier identifier, List partitionKeys, CatalogLoader catalogLoader) { + return new CatalogFormatTablePartitionManager(identifier, partitionKeys, catalogLoader); + } +} 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..44908749cad6 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 @@ -32,6 +32,7 @@ import org.apache.paimon.fs.Path; import org.apache.paimon.manifest.PartitionEntry; import org.apache.paimon.options.Options; +import org.apache.paimon.partition.Partition; import org.apache.paimon.partition.PartitionPredicate; import org.apache.paimon.partition.PartitionPredicate.AndPartitionPredicate; import org.apache.paimon.partition.PartitionPredicate.DefaultPartitionPredicate; @@ -59,12 +60,14 @@ import javax.annotation.Nullable; +import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.Comparator; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -81,9 +84,10 @@ public class FormatTableScan implements InnerTableScan { private static final Logger LOG = LoggerFactory.getLogger(FormatTableScan.class); - private final FormatTable table; - private final CoreOptions coreOptions; + final FormatTable table; + final CoreOptions coreOptions; @Nullable private PartitionPredicate partitionFilter; + @Nullable private final FormatTablePartitionManager partitionManager; @Nullable private final Integer limit; private final long targetSplitSize; private final long openFileCost; @@ -97,6 +101,7 @@ public FormatTableScan( this.coreOptions = new CoreOptions(table.options()); this.partitionFilter = partitionFilter; this.limit = limit; + this.partitionManager = table.partitionManager(); this.targetSplitSize = coreOptions.splitTargetSize(); this.openFileCost = coreOptions.splitOpenFileCost(); this.format = table.format(); @@ -115,6 +120,29 @@ public Plan plan() { @Override public List listPartitionEntries() { + if (partitionManager != null) { + List partitions = partitionManager.listPartitions(Collections.emptyMap()); + if (partitions.isEmpty()) { + warnIfFilesystemPartitionsExist(); + } + boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath(); + List entries = new ArrayList<>(partitions.size()); + Set> seen = new HashSet<>(partitions.size()); + for (Partition partition : partitions) { + if (!seen.add(partition.spec())) { + continue; + } + entries.add( + new PartitionEntry( + toPartitionRow(normalizeSpec(partition.spec(), onlyValueInPath)), + partition.recordCount(), + partition.fileSizeInBytes(), + partition.fileCount(), + partition.lastFileCreationTime(), + partition.totalBuckets())); + } + return entries; + } List, Path>> partition2Paths = searchPartSpecAndPaths( table.fileIO(), @@ -142,7 +170,7 @@ public static boolean isDataFileName(String fileName) { return fileName != null && !fileName.startsWith(".") && !fileName.startsWith("_"); } - private BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { + BinaryRow toPartitionRow(LinkedHashMap partitionSpec) { RowType partitionType = table.partitionType(); GenericRow row = convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName()); @@ -160,7 +188,28 @@ 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) { + if (partitionManager == null) { + throw e; + } + // A registered partition without a directory reads as empty, + // matching Hive (e.g. ADD PARTITION before the first INSERT). + // Warn so a directory removed behind the catalog's back stays + // discoverable. + LOG.warn( + "Partition '{}' of 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 " + + "MSCK REPAIR TABLE.", + PartitionPathUtils.generatePartitionName( + partitionSpec, false), + table.fullName(), + pair.getValue()); + } } } } else { @@ -178,6 +227,16 @@ public List splits() { } List, Path>> findPartitions() { + if (partitionManager != null) { + Map prefix = leadingEqualityPrefix(); + List partitions = partitionManager.listPartitions(prefix); + if (partitions.isEmpty() && prefix.isEmpty()) { + warnIfFilesystemPartitionsExist(); + } + // The prefix is a coarse pre-filter; the full predicate is applied per partition in + // the plan, as it is for filesystem discovery. + return toSpecsAndPaths(partitions, coreOptions.formatTablePartitionOnlyValueInPath()); + } LOG.debug( "Find partitions for format table {}, partition filter: {}", table.name(), @@ -222,6 +281,101 @@ List, Path>> findPartitions() { } } + /** + * Turn the partitions a catalog reports into the specs and directories to read. Catalog + * metadata is not trusted for path construction: a spec that cannot form a partition directory + * of this table is rejected rather than resolved to some other directory. + */ + private List, Path>> toSpecsAndPaths( + List partitions, boolean onlyValueInPath) { + 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 spec = normalizeSpec(partition.spec(), onlyValueInPath); + String partitionPath = + PartitionPathUtils.generatePartitionPathUtil(spec, onlyValueInPath); + if (seenPartitionPaths.add(partitionPath)) { + result.add(Pair.of(spec, new Path(tablePath, partitionPath))); + } + } + return result; + } + + /** Order a catalog spec by the table partition keys and check it can form a directory. */ + private LinkedHashMap normalizeSpec( + @Nullable Map spec, boolean onlyValueInPath) { + List partitionKeys = table.partitionKeys(); + if (spec == null + || spec.size() != partitionKeys.size() + || !spec.keySet().containsAll(partitionKeys)) { + throw corruptPartitionSpec(spec); + } + LinkedHashMap normalized = new LinkedHashMap<>(); + for (String partitionKey : partitionKeys) { + String value = spec.get(partitionKey); + // In a value-only layout, '.' and '..' are complete path components and would resolve + // to a directory outside 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 format table %s; " + + "expected exactly the partition keys %s with values usable as " + + "path components.", + spec, table.fullName(), table.partitionKeys())); + } + + /** + * The leading equality prefix of the partition predicate, in partition-key order, pushed down + * to the partition catalog. Empty when there is no predicate or it does not start with + * equalities on the leading partition keys. + */ + private Map leadingEqualityPrefix() { + Optional predicate = extractPartitionPredicate(partitionFilter); + if (!predicate.isPresent()) { + return Collections.emptyMap(); + } + return extractLeadingEqualityPartitionSpecWhenOnlyAnd( + table.partitionKeys(), predicate.get(), table.partitionType()); + } + + /** + * Warn when the catalog knows no partitions but the table directory contains subdirectories: + * typically a table that predates catalog-managed partitions (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( + "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 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. + } + } + protected static List, Path>> generatePartitions( List partitionKeys, RowType partitionType, 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..44ded39b6ab5 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,53 @@ public static LinkedHashMap extractPartitionSpecFromPath(Path cu return fullPartSpec; } + /** + * Extract exactly the trailing {@code key=value} components for the declared partition keys, or + * null when the path does not end in them. Values are unescaped, so they are the raw partition + * values. + */ + @Nullable + public 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; + } + + @Nullable public static LinkedHashMap extractPartitionSpecFromPathOnlyValue( Path currPath, List partitionKeys) { LinkedHashMap fullPartSpec = new LinkedHashMap<>(); String[] split = currPath.toString().split(Path.SEPARATOR); + if (split.length < partitionKeys.size()) { + return null; + } 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 +463,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 +494,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 +521,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 +556,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/CatalogUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java new file mode 100644 index 000000000000..3ffb6a1ecbde --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java @@ -0,0 +1,135 @@ +/* + * 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 testRejectEngineImplementationForCatalogManagedPartitionsOnCreate() { + 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(); + + // The combination is rejected where catalog-managed partitions are actually served. + assertThatThrownBy( + () -> + CatalogUtils.validateCatalogManagedFormatTablePartitions( + Identifier.create("db", "t"), schema.options(), false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()) + .hasMessageContaining(FORMAT_TABLE_IMPLEMENTATION.key()); + + // Elsewhere it is not this validation's business: the option is inert there. + validateCreateTable(schema, false); + } + + @Test + void testEngineImplementationCannotHaveCatalogManagedPartitions() { + // Asking for catalog-managed partitions where they cannot be served is an error, not a + // table that silently reads its partitions from somewhere else. + assertThatThrownBy(() -> loadFormatTableRequestingCatalogManagedPartitions("engine", true)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()) + .hasMessageContaining(FORMAT_TABLE_IMPLEMENTATION.key()); + } + + @Test + void testOutsideRestCatalogPartitionsFromCatalogAreRejected() { + // The option names one source only, so a catalog that cannot serve it says so rather than + // reading the table's partitions from somewhere the option did not ask for. + assertThatThrownBy(() -> loadFormatTableRequestingCatalogManagedPartitions("paimon", false)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()) + .hasMessageContaining("REST catalog"); + } + + @Test + void testLoadCatalogManagedPartitionsFromRestCatalog() throws Exception { + Table table = loadFormatTableRequestingCatalogManagedPartitions("paimon", true); + assertThat(table).isInstanceOf(FormatTable.class); + FormatTable formatTable = (FormatTable) table; + assertThat( + new org.apache.paimon.CoreOptions(formatTable.options()) + .partitionedTableInMetastore()) + .isTrue(); + assertThat(formatTable.partitionManager()).isNotNull(); + } + + private static Table loadFormatTableRequestingCatalogManagedPartitions( + String formatTableImplementation, boolean isRestCatalog) throws Exception { + Identifier identifier = + Identifier.create("catalog_partition_db", "catalog_partition_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/catalog_partition_db.db/catalog_partition_table") + .build(); + TableMetadata metadata = + new TableMetadata( + TableSchema.create(0, schema), false, "catalog-partition-table-id"); + Catalog catalog = mock(Catalog.class); + when(catalog.catalogLoader()).thenReturn(() -> catalog); + + return loadTable( + catalog, + identifier, + path -> new LocalFileIO(), + path -> new LocalFileIO(), + ignored -> metadata, + null, + null, + null, + isRestCatalog); + } +} 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..094ede3847ab 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,12 @@ 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.SchemaManager; import org.apache.paimon.types.DataTypes; import org.apache.paimon.shade.guava30.com.google.common.collect.Lists; @@ -77,4 +80,33 @@ public void testAlterDatabase() throws Exception { false)) .isInstanceOf(UnsupportedOperationException.class); } + + @Test + public void testPartitionsFromCatalogAreRejectedOutsideRestCatalog() throws Exception { + String database = "rest_partition_source_db"; + Identifier identifier = Identifier.create(database, "rest_partition_source_table"); + catalog.createDatabase(database, false); + // Write the schema file directly to simulate a format table carrying the REST-only + // partition source 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); + + // The option names one thing only, so a catalog that cannot serve it says so rather than + // reading the table's partitions from somewhere the option did not ask for. + assertThatThrownBy(() -> catalog.getTable(identifier)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(CoreOptions.METASTORE_PARTITIONED_TABLE.key()) + .hasMessageContaining("REST catalog"); + } } 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..6cc8f7042018 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,8 +48,11 @@ 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.FormatTablePartitionManager; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.InstantiationUtil; import org.apache.paimon.utils.JsonSerdeUtil; import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap; @@ -216,8 +221,8 @@ void testCreateTableDefaultOptions() throws Exception { } @Test - void testManagedFormatTablePagedPartitionListingDoesNotFallback() throws Exception { - Identifier identifier = createManagedFormatTable(); + void testCatalogManagedPagedPartitionListingDoesNotFallback() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); restCatalogServer.setPartitionListingSupported(false); assertThatThrownBy(() -> restCatalog.listPartitionsPaged(identifier, null, null, null)) @@ -225,8 +230,8 @@ void testManagedFormatTablePagedPartitionListingDoesNotFallback() throws Excepti } @Test - void testManagedFormatTablePartitionListingByNamesDoesNotFallback() throws Exception { - Identifier identifier = createManagedFormatTable(); + void testCatalogManagedPartitionListingByNamesDoesNotFallback() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); restCatalogServer.setPartitionListingSupported(false); assertThatThrownBy( @@ -239,15 +244,97 @@ void testManagedFormatTablePartitionListingByNamesDoesNotFallback() throws Excep } @Test - void testManagedFormatTablePartitionListingDoesNotFallback() throws Exception { - Identifier identifier = createManagedFormatTable(); + void testCatalogManagedPartitionListingDoesNotFallback() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); restCatalogServer.setPartitionListingSupported(false); assertThatThrownBy(() -> restCatalog.listPartitions(identifier)) .isInstanceOf(NotImplementedException.class); } - private Identifier createManagedFormatTable() throws Exception { + @Test + void testCatalogManagedPartitionListingReflectsCatalogMutationsImmediately() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); + FormatTable table = (FormatTable) restCatalog.getTable(identifier); + FormatTablePartitionManager partitionManager = table.partitionManager(); + assertThat(partitionManager).isNotNull(); + assertThat(partitionManager.listPartitions(Collections.emptyMap())).isEmpty(); + Map partition = Collections.singletonMap("dt", "20260717"); + + restCatalog.createPartitions(identifier, Collections.singletonList(partition)); + + // Listings are not cached, so a mutation through the catalog is visible to the next read. + assertThat(partitionManager.listPartitions(Collections.emptyMap())) + .extracting(org.apache.paimon.partition.Partition::spec) + .containsExactly(partition); + + restCatalog.dropPartitions(identifier, Collections.singletonList(partition)); + + assertThat(partitionManager.listPartitions(Collections.emptyMap())).isEmpty(); + } + + @Test + void testPartitionManagerSurvivesSerialization() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); + FormatTable table = (FormatTable) restCatalog.getTable(identifier); + Map partition = Collections.singletonMap("dt", "20260717"); + restCatalog.createPartitions(identifier, Collections.singletonList(partition)); + + // A table travels to task processes; its partition catalog must rebuild its client there. + FormatTablePartitionManager roundTripped = + InstantiationUtil.clone(table.partitionManager()); + + assertThat(roundTripped.listPartitions(Collections.emptyMap())) + .extracting(org.apache.paimon.partition.Partition::spec) + .containsExactly(partition); + } + + @Test + void testRejectCatalogManagedPartitionsOnExternalTableBeforeCreate() throws Exception { + Identifier identifier = Identifier.create("db1", "external_partitioned_format_table"); + restCatalog.createDatabase(identifier.getDatabaseName(), true); + String externalPath = dataPath + "/external-partitioned-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 testRoundTrippedFormatTableReplacePassesClientValidation() throws Exception { + Identifier identifier = createFormatTableWithCatalogManagedPartitions(); + 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 createFormatTableWithCatalogManagedPartitions() throws Exception { Identifier identifier = Identifier.create("db1", "managed_partition_table"); restCatalog.createDatabase(identifier.getDatabaseName(), true); restCatalog.createTable( 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..0e386e8b10cb 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; @@ -1535,8 +1536,8 @@ void testListPartitionsFromFile() throws Exception { } @Test - void testCreatePartitionsForManagedFormatTable() throws Exception { - Identifier identifier = Identifier.create("format_partition_db", "managed_table"); + void testCreatePartitionsForCatalogManagedFormatTablePartitions() throws Exception { + Identifier identifier = Identifier.create("format_partition_db", "catalog_partition_table"); catalog.createDatabase(identifier.getDatabaseName(), true); catalog.createTable( identifier, @@ -1576,8 +1577,9 @@ void testCreatePartitionsForManagedFormatTable() throws Exception { } @Test - void testDropPartitionsForManagedFormatTable() throws Exception { - Identifier identifier = Identifier.create("format_partition_db", "managed_drop_table"); + void testDropPartitionsForCatalogManagedFormatTablePartitions() throws Exception { + Identifier identifier = + Identifier.create("format_partition_db", "catalog_partition_drop_table"); catalog.createDatabase(identifier.getDatabaseName(), true); catalog.createTable( identifier, @@ -1639,6 +1641,70 @@ void testDropPartitionsLoadsNonManagedTableOnce() throws Exception { Mockito.verify(catalogSpy, Mockito.times(1)).getTable(identifier); } + @Test + void testCatalogManagedPartitionCommitAndScanMatchesFileSystemMode() throws Exception { + Identifier identifier = + Identifier.create("format_partition_db", "catalog_partition_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); + // The table carries the catalog partition metadata its scan reads from. + assertThat(managedTable.partitionManager()).isNotNull(); + + 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"); + List readFromCatalog = read(managedTable, null, null, partitionFilter, null); + // Assert the rows themselves first: comparing the two reads alone would also pass if + // both stopped returning anything. + assertThat(readFromCatalog) + .extracting(row -> row.getInt(0) + "," + row.getInt(1) + "," + row.getInt(2)) + .containsExactlyInAnyOrder("2,2025,10", "3,2025,11"); + assertThat(readFromCatalog) + .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..36064eea52f8 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java @@ -0,0 +1,174 @@ +/* + * 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.Identifier; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.table.format.FormatTablePartitionManager; +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.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 testCopyRejectsCatalogManagedPartitionStateChange(boolean withCatalogManagedPartitions) { + FormatTable table = formatTable(Boolean.toString(withCatalogManagedPartitions)); + + assertThatThrownBy( + () -> + table.copy( + Collections.singletonMap( + METASTORE_PARTITIONED_TABLE.key(), + Boolean.toString(!withCatalogManagedPartitions)))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key()); + } + + @Test + void testCopyAllowsSemanticallyEquivalentCatalogManagedPartitionOption() { + 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(), "true"); + } + + @Test + void testCopyRejectsChangingAbsentCatalogManagedPartitionDefault() { + 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 testCatalogManagedPartitionCopyRejectsPhysicalPartitionLayoutChange( + 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)); + // The layout is only fixed for a table that actually has catalog-managed partitions. + FormatTable table = + formatTable( + options, + FormatTablePartitionManager.create( + Identifier.create( + "catalog_partition_db", "catalog_partition_table"), + Collections.singletonList("dt"), + () -> 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 testCopyPreservesPartitionModeAndCatalogProviderForUnrelatedOption( + boolean withCatalogManagedPartitions) { + Identifier identifier = + Identifier.create("catalog_partition_db", "catalog_partition_table"); + FormatTablePartitionManager partitionManager = + FormatTablePartitionManager.create( + identifier, Collections.singletonList("dt"), () -> null); + Map options = new LinkedHashMap<>(); + options.put( + METASTORE_PARTITIONED_TABLE.key(), Boolean.toString(withCatalogManagedPartitions)); + FormatTable table = formatTable(options, partitionManager); + + 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(withCatalogManagedPartitions)); + assertThat( + new org.apache.paimon.CoreOptions(copied.options()) + .partitionedTableInMetastore()) + .isEqualTo(withCatalogManagedPartitions); + assertThat(copied.partitionManager()).isSameAs(partitionManager); + assertThat(table.options()) + .containsExactlyEntriesOf(options) + .doesNotContainKey(READ_BATCH_SIZE.key()); + } + + @Test + void testPartitionManagerIsOptionalForOtherImplementations() throws Exception { + // Implementations outside this repository do not have to know about catalog-managed + // partitions. + assertThat(FormatTable.class.getMethod("partitionManager").isDefault()).isTrue(); + } + + private static FormatTable formatTable(String partitionedInMetastore) { + return formatTable( + Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), partitionedInMetastore), + null); + } + + private static FormatTable formatTable( + Map options, FormatTablePartitionManager partitionManager) { + return FormatTable.builder() + .fileIO(LocalFileIO.create()) + .identifier(Identifier.create("catalog_partition_db", "catalog_partition_table")) + .rowType( + RowType.builder() + .field("id", DataTypes.INT()) + .field("dt", DataTypes.STRING()) + .build()) + .partitionKeys(Collections.singletonList("dt")) + .location("file:///warehouse/catalog_partition_table") + .format(FormatTable.Format.PARQUET) + .options(options) + .partitionManager(partitionManager) + .build(); + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java new file mode 100644 index 000000000000..d7a5d97e691a --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogFormatTablePartitionManagerTest.java @@ -0,0 +1,417 @@ +/* + * 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.CatalogLoader; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.partition.Partition; +import org.apache.paimon.utils.InstantiationUtil; + +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +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.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.catchThrowable; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * Tests for {@link CatalogFormatTablePartitionManager}: paging, prefix pushdown, per-request + * batching and catalog lifecycle are transport details this adapter owns, and none of them are + * visible through {@link FormatTablePartitionManager}. + */ +class CatalogFormatTablePartitionManagerTest { + + private static final Identifier IDENTIFIER = + Identifier.create("catalog_partition_db", "catalog_partition_table"); + + private static final List PARTITION_KEYS = Arrays.asList("year", "month"); + + /** One catalog request never carries more than this many partitions. */ + private static final int REQUEST_SIZE = 1000; + + /** Handed to the serializable loader, which cannot capture a test instance field. */ + private static Catalog staticCatalog; + + // ------------------------------------------------------------------------ + // paging + // ------------------------------------------------------------------------ + + @Test + void testListPartitionsFollowsEveryPage() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition first = partition("2025", "01"); + Partition second = partition("2025", "02"); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn( + new PagedList<>(Collections.singletonList(first), "page-2"), + new PagedList<>(Collections.singletonList(second), null)); + + List partitions = + partitionManager(catalog).listPartitions(Collections.emptyMap()); + + assertThat(partitions).containsExactly(first, second); + ArgumentCaptor pageTokens = ArgumentCaptor.forClass(String.class); + verify(catalog, times(2)) + .listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), pageTokens.capture(), any()); + assertThat(pageTokens.getAllValues()).containsExactly(null, "page-2"); + } + + @Test + void testEmptyNextPageTokenEndsPaging() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition only = partition("2025", "01"); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.singletonList(only), "")); + + List partitions = + partitionManager(catalog).listPartitions(Collections.emptyMap()); + + assertThat(partitions).containsExactly(only); + verify(catalog, times(1)).listPartitionsPaged(any(), any(), any(), any()); + } + + @Test + void testRepeatedPageTokenFailsFast() throws Exception { + Catalog catalog = mock(Catalog.class); + // a -> b -> a: following the cycle would list the same pages forever. + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn( + new PagedList<>(Collections.singletonList(partition("2025", "01")), "a"), + new PagedList<>(Collections.singletonList(partition("2025", "02")), "b"), + new PagedList<>(Collections.singletonList(partition("2025", "03")), "a")); + + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap())) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("repeated partition page token") + .hasMessageContaining("catalog_partition_db.catalog_partition_table"); + } + + // ------------------------------------------------------------------------ + // prefix pushdown and client side filtering + // ------------------------------------------------------------------------ + + @Test + void testLeadingPrefixIsPushedDownAsPattern() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.emptyList(), null)); + + partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025")); + + verify(catalog) + .listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), isNull(), eq("year=2025/%")); + } + + @Test + void testEmptyPrefixPushesDownNoPattern() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.emptyList(), null)); + + partitionManager(catalog).listPartitions(Collections.emptyMap()); + + verify(catalog).listPartitionsPaged(eq(IDENTIFIER), eq(REQUEST_SIZE), isNull(), isNull()); + } + + @Test + void testPrefixIsEnforcedOnCatalogsThatIgnoreThePattern() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition matching = partition("2025", "01"); + Partition other = partition("2024", "01"); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Arrays.asList(matching, other), null)); + + List partitions = + partitionManager(catalog).listPartitions(Collections.singletonMap("year", "2025")); + + assertThat(partitions).containsExactly(matching); + } + + @Test + void testNonLeadingPrefixIsRejected() { + Catalog catalog = mock(Catalog.class); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + // "month" without "year" cannot be expressed as a partition path prefix. + Map prefix = Collections.singletonMap("month", "10"); + + assertThatThrownBy(() -> partitionManager.listPartitions(prefix)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("leading prefix"); + } + + // ------------------------------------------------------------------------ + // batching + // ------------------------------------------------------------------------ + + @Test + void testIdempotentCreateIsSplitIntoRequests() throws Exception { + Catalog catalog = mock(Catalog.class); + List> specs = specs(2500); + + partitionManager(catalog).createPartitions(specs, true); + + List>> batches = capturedCreates(catalog, true, 3); + assertThat(batches).extracting(List::size).containsExactly(1000, 1000, 500); + assertThat(flatten(batches)).isEqualTo(specs); + } + + @Test + void testStrictCreateStaysOneRequest() throws Exception { + Catalog catalog = mock(Catalog.class); + List> specs = specs(2500); + + partitionManager(catalog).createPartitions(specs, false); + + // Rejecting the whole batch when any partition exists only holds if it is never split. + List>> batches = capturedCreates(catalog, false, 1); + assertThat(batches.get(0)).isEqualTo(specs); + } + + @Test + void testDropIsSplitIntoRequests() throws Exception { + Catalog catalog = mock(Catalog.class); + List> specs = specs(2500); + + partitionManager(catalog).dropPartitions(specs); + + @SuppressWarnings("unchecked") + ArgumentCaptor>> captor = ArgumentCaptor.forClass(List.class); + verify(catalog, times(3)).dropPartitions(eq(IDENTIFIER), captor.capture()); + assertThat(captor.getAllValues()).extracting(List::size).containsExactly(1000, 1000, 500); + assertThat(flatten(captor.getAllValues())).isEqualTo(specs); + } + + @Test + void testListByNamesIsSplitAndUnioned() throws Exception { + Catalog catalog = mock(Catalog.class); + List> specs = specs(2500); + Partition first = partition("2025", "0000"); + Partition second = partition("2025", "1000"); + Partition third = partition("2025", "2000"); + when(catalog.listPartitionsByNames(any(), anyList())) + .thenReturn( + Collections.singletonList(first), + Collections.singletonList(second), + Collections.singletonList(third)); + + List found = partitionManager(catalog).listPartitionsByNames(specs); + + assertThat(found).containsExactly(first, second, third); + @SuppressWarnings("unchecked") + ArgumentCaptor>> captor = ArgumentCaptor.forClass(List.class); + verify(catalog, times(3)).listPartitionsByNames(eq(IDENTIFIER), captor.capture()); + assertThat(captor.getAllValues()).extracting(List::size).containsExactly(1000, 1000, 500); + assertThat(flatten(captor.getAllValues())).isEqualTo(specs); + } + + @Test + void testEmptyInputTouchesNoCatalog() { + Catalog createCatalog = mock(Catalog.class); + partitionManager(createCatalog).createPartitions(Collections.emptyList(), true); + verifyNoInteractions(createCatalog); + + Catalog dropCatalog = mock(Catalog.class); + partitionManager(dropCatalog).dropPartitions(Collections.emptyList()); + verifyNoInteractions(dropCatalog); + + Catalog listCatalog = mock(Catalog.class); + assertThat(partitionManager(listCatalog).listPartitionsByNames(Collections.emptyList())) + .isEmpty(); + verifyNoInteractions(listCatalog); + } + + // ------------------------------------------------------------------------ + // catalog lifecycle + // ------------------------------------------------------------------------ + + @Test + void testEveryOperationClosesTheCatalog() throws Exception { + Catalog listCatalog = mock(Catalog.class); + when(listCatalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.emptyList(), null)); + partitionManager(listCatalog).listPartitions(Collections.emptyMap()); + verify(listCatalog).close(); + + Catalog byNamesCatalog = mock(Catalog.class); + when(byNamesCatalog.listPartitionsByNames(any(), anyList())) + .thenReturn(Collections.emptyList()); + partitionManager(byNamesCatalog).listPartitionsByNames(specs(1)); + verify(byNamesCatalog).close(); + + Catalog createCatalog = mock(Catalog.class); + partitionManager(createCatalog).createPartitions(specs(1), true); + verify(createCatalog).close(); + + Catalog dropCatalog = mock(Catalog.class); + partitionManager(dropCatalog).dropPartitions(specs(1)); + verify(dropCatalog).close(); + } + + @Test + void testFailingOperationStillClosesTheCatalog() throws Exception { + Catalog catalog = mock(Catalog.class); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenThrow(new RuntimeException("catalog unavailable")); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + assertThatThrownBy(() -> partitionManager.listPartitions(Collections.emptyMap())) + .isInstanceOf(RuntimeException.class); + + verify(catalog).close(); + } + + @Test + void testRejectedPrefixNeverLoadsACatalog() { + Catalog catalog = mock(Catalog.class); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + Map prefix = Collections.singletonMap("month", "10"); + + assertThatThrownBy(() -> partitionManager.listPartitions(prefix)) + .isInstanceOf(IllegalArgumentException.class); + + verifyNoInteractions(catalog); + } + + // ------------------------------------------------------------------------ + // error translation + // ------------------------------------------------------------------------ + + @Test + void testCheckedExceptionIsWrappedWithTableContext() throws Exception { + Catalog catalog = mock(Catalog.class); + Catalog.TableNotExistException notExist = new Catalog.TableNotExistException(IDENTIFIER); + when(catalog.listPartitionsPaged(any(), any(), any(), any())).thenThrow(notExist); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + Throwable thrown = + catchThrowable(() -> partitionManager.listPartitions(Collections.emptyMap())); + + assertThat(thrown) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("list partitions") + .hasMessageContaining("catalog_partition_db.catalog_partition_table"); + assertThat(thrown.getCause()).isSameAs(notExist); + } + + @Test + void testRuntimeExceptionIsRethrownUnchanged() throws Exception { + Catalog catalog = mock(Catalog.class); + RuntimeException failure = new IllegalStateException("catalog unavailable"); + doThrow(failure).when(catalog).createPartitions(any(), anyList(), anyBoolean()); + FormatTablePartitionManager partitionManager = partitionManager(catalog); + + Throwable thrown = catchThrowable(() -> partitionManager.createPartitions(specs(1), true)); + + assertThat(thrown).isSameAs(failure); + } + + // ------------------------------------------------------------------------ + // serializability + // ------------------------------------------------------------------------ + + @Test + void testSurvivesJavaSerialization() throws Exception { + Catalog catalog = mock(Catalog.class); + Partition only = partition("2025", "01"); + when(catalog.listPartitionsPaged(any(), any(), any(), any())) + .thenReturn(new PagedList<>(Collections.singletonList(only), null)); + staticCatalog = catalog; + + FormatTablePartitionManager partitionManager = + FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS, new TestLoader()); + FormatTablePartitionManager cloned = InstantiationUtil.clone(partitionManager); + + assertThat(cloned.listPartitions(Collections.emptyMap())).containsExactly(only); + verify(catalog).close(); + } + + // ------------------------------------------------------------------------ + // helpers + // ------------------------------------------------------------------------ + + private static FormatTablePartitionManager partitionManager(Catalog catalog) { + return FormatTablePartitionManager.create(IDENTIFIER, PARTITION_KEYS, () -> catalog); + } + + private static List>> capturedCreates( + Catalog catalog, boolean ignoreIfExists, int expectedRequests) throws Exception { + @SuppressWarnings("unchecked") + ArgumentCaptor>> captor = ArgumentCaptor.forClass(List.class); + verify(catalog, times(expectedRequests)) + .createPartitions(eq(IDENTIFIER), captor.capture(), eq(ignoreIfExists)); + return captor.getAllValues(); + } + + private static List> flatten(List>> batches) { + return batches.stream().flatMap(List::stream).collect(Collectors.toList()); + } + + private static Partition partition(String year, String month) { + return new Partition(spec(year, month), 0, 0, 0, 0, -1, false); + } + + private static Map spec(String year, String month) { + LinkedHashMap spec = new LinkedHashMap<>(); + spec.put("year", year); + spec.put("month", month); + return spec; + } + + private static List> specs(int count) { + List> specs = new ArrayList<>(count); + for (int i = 0; i < count; i++) { + specs.add(spec("2025", String.format("%04d", i))); + } + return specs; + } + + /** A {@link CatalogLoader} that survives serialization by holding no state of its own. */ + private static class TestLoader implements CatalogLoader { + + private static final long serialVersionUID = 1L; + + @Override + public Catalog load() { + return staticCatalog; + } + } +} diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java new file mode 100644 index 000000000000..da74ea32732f --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/table/format/CatalogManagedPartitionScanTest.java @@ -0,0 +1,451 @@ +/* + * 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.mock; +import static org.mockito.Mockito.when; + +/** Tests for Format Table scans whose partitions are catalog-managed. */ +class CatalogManagedPartitionScanTest { + + private static final Identifier IDENTIFIER = + Identifier.create("catalog_partition_db", "catalog_partition_table"); + + @TempDir java.nio.file.Path tempDir; + + @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, partitionManager(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); + + FormatTableScan scan = new FormatTableScan(table, filter, null); + List plannedFiles = plannedFiles(scan.plan().splits()); + + 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, partitionManager(catalog), false); + + List entries = + new FormatTableScan(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, recordingCatalog(Collections.emptyList())); + Predicate predicate = + new PredicateBuilder(table.partitionType()) + .equal(0, BinaryString.fromString("a_b")); + PartitionPredicate filter = + PartitionPredicate.fromPredicate(table.partitionType(), predicate); + + List plannedFiles = + plannedFiles(new FormatTableScan(table, filter, null).plan().splits()); + + // '_' has no special meaning in the prefix contract; it must not widen the match. + assertThat(plannedFiles).isEmpty(); + assertThat(requestedPrefixes).containsExactly(Collections.singletonMap("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, partitionManager(catalog), true); + + List plannedFiles = + plannedFiles(new FormatTableScan(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, partitionManager(catalog), false); + + List plannedFiles = + plannedFiles(new FormatTableScan(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, partitionManager(catalog)); + + List plannedFiles = + plannedFiles(new FormatTableScan(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, partitionManager(catalog)); + + assertThatThrownBy(() -> new FormatTableScan(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, partitionManager(catalog), false); + + assertThatThrownBy(() -> new FormatTableScan(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 testMissingCatalogRegisteredPartitionReadsAsEmpty() 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, partitionManager(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 FormatTableScan(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, partitionManager(catalog), true); + + assertThatThrownBy(() -> new FormatTableScan(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, partitionManager(catalog)); + + List plannedFiles = + plannedFiles(new FormatTableScan(table, null, null).plan().splits()); + + assertThat(plannedFiles).containsExactly(dataFile); + } + + private final List> requestedPrefixes = new ArrayList<>(); + + private FormatTablePartitionManager partitionManager(Catalog catalog) { + return FormatTablePartitionManager.create( + IDENTIFIER, Arrays.asList("year", "month"), () -> catalog); + } + + /** Records the prefix the scan pushes down and answers from a fixed partition list. */ + private FormatTablePartitionManager recordingCatalog(List partitions) { + List> prefixes = requestedPrefixes; + return new FormatTablePartitionManager() { + @Override + public List listPartitions(Map prefix) { + prefixes.add(new LinkedHashMap<>(prefix)); + List matching = new ArrayList<>(); + for (Partition partition : partitions) { + boolean matches = true; + for (Map.Entry entry : prefix.entrySet()) { + if (!entry.getValue().equals(partition.spec().get(entry.getKey()))) { + matches = false; + break; + } + } + if (matches) { + matching.add(partition); + } + } + return matching; + } + + @Override + public List listPartitionsByNames(List> partitions) { + throw new UnsupportedOperationException(); + } + + @Override + public void createPartitions( + List> partitions, boolean ignoreIfExists) { + throw new UnsupportedOperationException(); + } + + @Override + public void dropPartitions(List> partitions) { + throw new UnsupportedOperationException(); + } + }; + } + + private FormatTable createTable( + LocalFileIO fileIO, + Path tablePath, + FormatTablePartitionManager partitionManager, + 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))) + .partitionManager(partitionManager) + .build(); + } + + private FormatTable createStringPartitionTable( + LocalFileIO fileIO, Path tablePath, FormatTablePartitionManager partitionManager) { + return createStringPartitionTable(fileIO, tablePath, partitionManager, false); + } + + private FormatTable createStringPartitionTable( + LocalFileIO fileIO, + Path tablePath, + FormatTablePartitionManager partitionManager, + 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))) + .partitionManager(partitionManager) + .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/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java new file mode 100644 index 000000000000..c7f8c8dac1af --- /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.ArgumentMatchers.eq; +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 testPartitionRegistrationFailureDiscardsTheFilesItWrote() 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(); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + RuntimeException registrationFailure = + new RuntimeException("Catalog partition registration unavailable"); + doThrow(registrationFailure).when(partitionManager).createPartitions(anyList(), eq(true)); + + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + false, + Identifier.create("catalog_partition_db", "catalog_partition_table"), + null, + null, + null, + partitionManager); + CommitMessage message = new TwoPhaseCommitMessage(committer); + + assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) + .isInstanceOf(RuntimeException.class) + .hasRootCauseMessage("Catalog partition registration unavailable"); + + // A failed write leaves nothing behind, whichever step failed: rerunning it converges, + // and an idempotent registration makes a partition that was registered anyway harmless. + assertThat(fileIO.exists(targetPath)).isFalse(); + verify(partitionManager).createPartitions(anyList(), eq(true)); + } + + @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); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + FormatTableCommit commit = + new FormatTableCommit( + tablePath.toString(), + Arrays.asList("year", "month"), + fileIO, + false, + false, + Identifier.create("catalog_partition_db", "catalog_partition_table"), + null, + null, + null, + partitionManager); + CommitMessage message = new TwoPhaseCommitMessage(committer); + + assertThatThrownBy(() -> commit.commit(Collections.singletonList(message))) + .isInstanceOf(RuntimeException.class) + .hasRootCauseMessage("data commit failed"); + + verify(committer).discard(fileIO); + verify(partitionManager, never()).createPartitions(anyList(), eq(true)); + } + + @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/"); + + FormatTablePartitionManager partitionManager = + commitPartitionedFile(tablePath, false, partitionDir); + + // The catalog must receive RAW values; readers re-escape them when probing directories. + assertThat(registeredSpec(partitionManager)) + .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"); + + FormatTablePartitionManager partitionManager = + commitPartitionedFile(tablePath, false, "year=2025/month=10"); + + assertThat(registeredSpec(partitionManager)) + .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/"); + + FormatTablePartitionManager partitionManager = + commitPartitionedFile(tablePath, true, partitionDir); + + assertThat(registeredSpec(partitionManager)) + .containsExactly(entry("year", "2025"), entry("month", "a:b")); + } + + @Test + void testPathNotMatchingThePartitionKeysFails() { + Path tablePath = new Path(tempDir.toUri()); + + // The message names the path and the declared keys, which is what tells a reader that + // 'day' is not where 'month' was expected. + assertThatThrownBy(() -> commitPartitionedFile(tablePath, false, "year=2025/day=10")) + .isInstanceOf(RuntimeException.class) + .hasRootCauseInstanceOf(IllegalArgumentException.class) + .rootCause() + .hasMessageContaining("year=2025/day=10") + .hasMessageContaining("catalog_partition_db.catalog_partition_table") + .hasMessageContaining("[year, month]"); + } + + @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("catalog_partition_db", "catalog_partition_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 FormatTablePartitionManager 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(); + FormatTablePartitionManager partitionManager = mock(FormatTablePartitionManager.class); + FormatTableCommit commit = + new FormatTableCommit( + tableLocation.toString(), + Arrays.asList("year", "month"), + fileIO, + onlyValueInPath, + false, + Identifier.create("catalog_partition_db", "catalog_partition_table"), + null, + null, + null, + partitionManager); + commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer))); + return partitionManager; + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private static Map registeredSpec( + FormatTablePartitionManager partitionManager) { + ArgumentCaptor>> captor = + ArgumentCaptor.forClass((Class) List.class); + verify(partitionManager).createPartitions(captor.capture(), eq(true)); + assertThat(captor.getValue()).hasSize(1); + return captor.getValue().get(0); + } +} 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();