> part
throw new TableNoPermissionException(identifier, e);
} catch (BadRequestException e) {
throw new IllegalArgumentException(e.getMessage());
+ } finally {
+ // The server may have committed even when the response was lost or malformed.
+ FormatTableCatalogProvider.advanceGeneration(identifier);
}
return;
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
index 9a832176caaf..64325e70b7c8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java
@@ -18,9 +18,11 @@
package org.apache.paimon.table;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.CatalogUtils;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.IndexManifestEntry;
@@ -29,6 +31,7 @@
import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.format.FormatBatchWriteBuilder;
import org.apache.paimon.table.format.FormatReadBuilder;
+import org.apache.paimon.table.format.FormatTableCatalogProvider;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.source.BatchVectorSearchBuilder;
@@ -56,8 +59,9 @@
* operations on this table allow for reading or writing to these files, facilitating the retrieval
* of existing data and the addition of new files.
*
- * Partitioned file format table just like the standard hive format. Partitions are discovered
- * and inferred based on directory structure.
+ *
A partitioned file format table uses the standard Hive directory layout. By default,
+ * partitions are discovered from that layout. Catalog-managed format tables use catalog metadata
+ * for partition visibility while retaining the same physical layout.
*
* @since 0.9.0
*/
@@ -75,6 +79,12 @@ public interface FormatTable extends Table {
CatalogContext catalogContext();
+ /** Catalog access used by managed partition discovery and registration. */
+ @Nullable
+ default FormatTableCatalogProvider catalogProvider() {
+ return null;
+ }
+
/** Currently supported formats. */
enum Format {
ORC,
@@ -115,6 +125,7 @@ class Builder {
private Map options;
@Nullable private String comment;
private CatalogContext catalogContext;
+ @Nullable private FormatTableCatalogProvider catalogProvider;
public Builder fileIO(FileIO fileIO) {
this.fileIO = fileIO;
@@ -161,6 +172,11 @@ public Builder catalogContext(CatalogContext catalogContext) {
return this;
}
+ public Builder catalogProvider(@Nullable FormatTableCatalogProvider catalogProvider) {
+ this.catalogProvider = catalogProvider;
+ return this;
+ }
+
public FormatTable build() {
return new FormatTableImpl(
fileIO,
@@ -171,7 +187,8 @@ public FormatTable build() {
format,
options,
comment,
- catalogContext);
+ catalogContext,
+ catalogProvider);
}
}
@@ -189,6 +206,7 @@ class FormatTableImpl implements FormatTable {
private final Map options;
@Nullable private final String comment;
private CatalogContext catalogContext;
+ @Nullable private final FormatTableCatalogProvider catalogProvider;
public FormatTableImpl(
FileIO fileIO,
@@ -200,6 +218,30 @@ public FormatTableImpl(
Map options,
@Nullable String comment,
CatalogContext catalogContext) {
+ this(
+ fileIO,
+ identifier,
+ rowType,
+ partitionKeys,
+ location,
+ format,
+ options,
+ comment,
+ catalogContext,
+ null);
+ }
+
+ public FormatTableImpl(
+ FileIO fileIO,
+ Identifier identifier,
+ RowType rowType,
+ List partitionKeys,
+ String location,
+ Format format,
+ Map options,
+ @Nullable String comment,
+ CatalogContext catalogContext,
+ @Nullable FormatTableCatalogProvider catalogProvider) {
this.fileIO = fileIO;
this.identifier = identifier;
this.rowType = rowType;
@@ -209,6 +251,7 @@ public FormatTableImpl(
this.options = options;
this.comment = comment;
this.catalogContext = catalogContext;
+ this.catalogProvider = catalogProvider;
}
@Override
@@ -265,6 +308,27 @@ public FileIO fileIO() {
public FormatTable copy(Map dynamicOptions) {
Map newOptions = new HashMap<>(options);
newOptions.putAll(dynamicOptions);
+
+ CoreOptions coreOptions = CoreOptions.fromMap(options);
+ CoreOptions copiedCoreOptions = CoreOptions.fromMap(newOptions);
+ boolean managed = coreOptions.partitionedTableInMetastore();
+ boolean copiedManaged = copiedCoreOptions.partitionedTableInMetastore();
+ if (managed != copiedManaged) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Dynamic option '%s' cannot change whether Format Table partitions are catalog-managed.",
+ CoreOptions.METASTORE_PARTITIONED_TABLE.key()));
+ }
+ if (managed
+ && coreOptions.formatTablePartitionOnlyValueInPath()
+ != copiedCoreOptions.formatTablePartitionOnlyValueInPath()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Dynamic option '%s' cannot change the physical partition layout of a catalog-managed Format Table.",
+ CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key()));
+ }
+ CatalogUtils.validateManagedFormatTableOptions(newOptions);
+
return new FormatTableImpl(
fileIO,
identifier,
@@ -274,7 +338,8 @@ public FormatTable copy(Map dynamicOptions) {
format,
newOptions,
comment,
- catalogContext);
+ catalogContext,
+ catalogProvider);
}
@Override
@@ -302,6 +367,12 @@ public FullTextSearchBuilder newFullTextSearchBuilder() {
public CatalogContext catalogContext() {
return this.catalogContext;
}
+
+ @Override
+ @Nullable
+ public FormatTableCatalogProvider catalogProvider() {
+ return catalogProvider;
+ }
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
index 924e67dbdfdc..d3970bb878ea 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatBatchWriteBuilder.java
@@ -87,7 +87,8 @@ public BatchTableCommit newCommit() {
Identifier.fromString(table.fullName()),
staticPartition,
syncHiveUri,
- table.catalogContext());
+ table.catalogContext(),
+ table.catalogProvider());
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
index 30a567b436a6..1c9010d0be04 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatReadBuilder.java
@@ -173,6 +173,16 @@ public TableScan newScan() {
partitionFilter = partitionPredicateOpt.get();
}
}
+ if (options.partitionedTableInMetastore()) {
+ if (table.catalogProvider() == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Managed format table %s has no catalog partition provider. "
+ + "The catalog client does not support managed format tables.",
+ table.fullName()));
+ }
+ return new ManagedFormatTableScan(table, partitionFilter, limit);
+ }
return new FormatTableScan(table, partitionFilter, limit);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java
new file mode 100644
index 000000000000..c297ef63e094
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCatalogProvider.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.PagedList;
+import org.apache.paimon.annotation.Experimental;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.CatalogLoader;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache;
+import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Serializable catalog access for a managed format table. */
+@Experimental
+public class FormatTableCatalogProvider implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private static final int PARTITION_PAGE_SIZE = 1000;
+ private static final int MAX_CACHED_PATTERNS = 128;
+ private static final int MAX_TRACKED_TABLE_GENERATIONS = 10_000;
+ private static final Duration PARTITION_CACHE_TTL = Duration.ofSeconds(30);
+ private static final Duration TABLE_GENERATION_TTL = Duration.ofHours(1);
+
+ // Keyed by identifier full name: listings themselves live in per-instance caches, so sharing
+ // a generation counter across table incarnations (or across catalogs using the same name) can
+ // only cause an extra invalidation, never a stale read.
+ private static final Cache GENERATIONS =
+ Caffeine.newBuilder()
+ .expireAfterAccess(TABLE_GENERATION_TTL)
+ .maximumSize(MAX_TRACKED_TABLE_GENERATIONS)
+ .executor(Runnable::run)
+ .build();
+
+ private final Identifier identifier;
+ private final CatalogLoader catalogLoader;
+
+ @Nullable private transient Cache> partitionCache;
+ // Reused across list/create calls: constructing a Catalog (HTTP client, auth provider, and
+ // for some auth providers an initial token fetch) per partition operation is expensive, and
+ // RESTCatalog.close() is a no-op so a long-lived instance leaks nothing. Recreated lazily
+ // after deserialization.
+ @Nullable private transient Catalog catalog;
+
+ public FormatTableCatalogProvider(Identifier identifier, CatalogLoader catalogLoader) {
+ this.identifier = identifier;
+ this.catalogLoader = catalogLoader;
+ }
+
+ /** List every catalog-visible partition matching the optional name prefix pattern. */
+ public List listPartitions(@Nullable String partitionNamePattern) {
+ return cache().get(
+ partitionCacheKey(partitionNamePattern),
+ ignored -> loadPartitions(partitionNamePattern));
+ }
+
+ /**
+ * Advance the partition-listing generation of every provider of this table in this JVM, so
+ * same-process scans skip cached listings taken before the mutation. The catalog partition DDL
+ * path (create/dropPartitions) calls this from a {@code finally} block: the mutation's outcome
+ * can be ambiguous (the server may have committed even when the response was lost), so the
+ * cached listing must be dropped after every attempt, not only after a confirmed success.
+ */
+ public static void advanceGeneration(Identifier identifier) {
+ GENERATIONS.get(identifier.getFullName(), ignored -> new AtomicLong()).incrementAndGet();
+ }
+
+ /** Create partitions with the catalog's idempotent create contract. */
+ public void createPartitions(List> partitions) {
+ if (partitions.isEmpty()) {
+ return;
+ }
+ try {
+ // Register in bounded batches: one backfill commit can touch tens of thousands of
+ // partitions (e.g. years of hourly data in a single dynamic-partition INSERT), and a
+ // single unbounded request either monopolizes a catalog worker or trips server-side
+ // request caps, failing the job after its data files are already committed.
+ // Registration is an idempotent ADD-only upsert, so a mid-way failure leaves a state
+ // that a rerun or MSCK converges from.
+ for (int start = 0; start < partitions.size(); start += PARTITION_PAGE_SIZE) {
+ catalog()
+ .createPartitions(
+ identifier,
+ partitions.subList(
+ start,
+ Math.min(start + PARTITION_PAGE_SIZE, partitions.size())));
+ }
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to register partitions for managed format table %s.",
+ identifier),
+ e);
+ } finally {
+ advanceGeneration(identifier);
+ cache().invalidateAll();
+ }
+ }
+
+ private List loadPartitions(@Nullable String partitionNamePattern) {
+ List partitions = new ArrayList<>();
+ Set seenPageTokens = new HashSet<>();
+ try {
+ Catalog catalog = catalog();
+ String pageToken = null;
+ do {
+ PagedList page =
+ catalog.listPartitionsPaged(
+ identifier, PARTITION_PAGE_SIZE, pageToken, partitionNamePattern);
+ partitions.addAll(page.getElements());
+ pageToken = page.getNextPageToken();
+ if (StringUtils.isNotEmpty(pageToken) && !seenPageTokens.add(pageToken)) {
+ throw new IllegalStateException(
+ String.format(
+ "Catalog returned repeated partition page token '%s' for managed format table %s.",
+ pageToken, identifier.getFullName()));
+ }
+ } while (StringUtils.isNotEmpty(pageToken));
+ return Collections.unmodifiableList(partitions);
+ } catch (RuntimeException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new RuntimeException(
+ String.format(
+ "Failed to list partitions for managed format table %s.", identifier),
+ e);
+ }
+ }
+
+ private synchronized Catalog catalog() {
+ if (catalog == null) {
+ catalog = catalogLoader.load();
+ }
+ return catalog;
+ }
+
+ private synchronized Cache> cache() {
+ if (partitionCache == null) {
+ partitionCache =
+ Caffeine.newBuilder()
+ .expireAfterWrite(PARTITION_CACHE_TTL)
+ .maximumSize(MAX_CACHED_PATTERNS)
+ .executor(Runnable::run)
+ .build();
+ }
+ return partitionCache;
+ }
+
+ private String partitionCacheKey(@Nullable String partitionNamePattern) {
+ // Partition mutations advance a process-local generation so providers in this JVM skip
+ // stale entries. Providers in other JVMs converge when PARTITION_CACHE_TTL expires.
+ return generation().get() + "\000" + partitionNamePattern;
+ }
+
+ private AtomicLong generation() {
+ return GENERATIONS.get(identifier.getFullName(), ignored -> new AtomicLong());
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
index ed98c6ba50ff..8d38388907c5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableCommit.java
@@ -48,7 +48,6 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.StringJoiner;
import static org.apache.paimon.table.format.FormatBatchWriteBuilder.validateStaticPartition;
@@ -63,6 +62,10 @@ public class FormatTableCommit implements BatchTableCommit {
protected boolean overwrite = false;
private Catalog hiveCatalog;
private Identifier tableIdentifier;
+ @Nullable private final FormatTableCatalogProvider catalogProvider;
+ // Volatile: commit() and abort() may run on different threads when callers drive the API
+ // directly; abort must observe that registration already began to keep its no-op guarantee.
+ private volatile boolean partitionRegistrationStarted;
public FormatTableCommit(
String location,
@@ -74,6 +77,30 @@ public FormatTableCommit(
@Nullable Map staticPartitions,
@Nullable String syncHiveUri,
CatalogContext catalogContext) {
+ this(
+ location,
+ partitionKeys,
+ fileIO,
+ formatTablePartitionOnlyValueInPath,
+ overwrite,
+ tableIdentifier,
+ staticPartitions,
+ syncHiveUri,
+ catalogContext,
+ null);
+ }
+
+ public FormatTableCommit(
+ String location,
+ List partitionKeys,
+ FileIO fileIO,
+ boolean formatTablePartitionOnlyValueInPath,
+ boolean overwrite,
+ Identifier tableIdentifier,
+ @Nullable Map staticPartitions,
+ @Nullable String syncHiveUri,
+ CatalogContext catalogContext,
+ @Nullable FormatTableCatalogProvider catalogProvider) {
this.location = location;
this.fileIO = fileIO;
this.formatTablePartitionOnlyValueInPath = formatTablePartitionOnlyValueInPath;
@@ -82,6 +109,7 @@ public FormatTableCommit(
this.overwrite = overwrite;
this.partitionKeys = partitionKeys;
this.tableIdentifier = tableIdentifier;
+ this.catalogProvider = catalogProvider;
if (syncHiveUri != null) {
try {
Options options = new Options();
@@ -143,7 +171,9 @@ public void commit(List commitMessages) {
for (TwoPhaseOutputStream.Committer committer : committers) {
committer.commit(this.fileIO);
- if (partitionKeys != null && !partitionKeys.isEmpty() && hiveCatalog != null) {
+ if (partitionKeys != null
+ && !partitionKeys.isEmpty()
+ && (hiveCatalog != null || catalogProvider != null)) {
partitionSpecs.add(
extractPartitionSpecFromPath(
committer.targetPath().getParent(), partitionKeys));
@@ -152,6 +182,14 @@ public void commit(List commitMessages) {
for (TwoPhaseOutputStream.Committer committer : committers) {
committer.clean(this.fileIO);
}
+ if (catalogProvider != null && !partitionSpecs.isEmpty()) {
+ partitionRegistrationStarted = true;
+ try {
+ catalogProvider.createPartitions(new ArrayList<>(partitionSpecs));
+ } catch (RuntimeException e) {
+ throw partitionRegistrationFailure(e);
+ }
+ }
for (Map partitionSpec : partitionSpecs) {
if (hiveCatalog != null) {
try {
@@ -172,11 +210,28 @@ public void commit(List commitMessages) {
}
} catch (Exception e) {
- this.abort(commitMessages);
- throw new RuntimeException(e);
+ if (!partitionRegistrationStarted) {
+ this.abort(commitMessages);
+ throw new RuntimeException(e);
+ }
+ throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e);
}
}
+ private RuntimeException partitionRegistrationFailure(RuntimeException cause) {
+ String tableName = tableIdentifier.getFullName();
+ return new RuntimeException(
+ String.format(
+ "Managed partition registration failed for %s after data files were "
+ + "committed. Committed data files were preserved because the "
+ + "catalog state may be ambiguous. Verify the catalog partition "
+ + "metadata and, if partitions are missing, re-run a partition "
+ + "metadata sync for %s (e.g. the Spark procedure "
+ + "sys.sync_format_table_metadata or MSCK REPAIR TABLE).",
+ tableName, tableName),
+ cause);
+ }
+
private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException {
Method hiveCreatePartitionsInHmsMethod =
hiveCatalog
@@ -192,12 +247,68 @@ private Method getHiveCreatePartitionsInHmsMethod() throws NoSuchMethodException
private LinkedHashMap extractPartitionSpecFromPath(
Path partitionPath, List partitionKeys) {
- if (formatTablePartitionOnlyValueInPath) {
- return PartitionPathUtils.extractPartitionSpecFromPathOnlyValue(
- partitionPath, partitionKeys);
- } else {
- return PartitionPathUtils.extractPartitionSpecFromPath(partitionPath);
+ // The writer always lays partitions out as //.../, so the partition
+ // spec is exactly the trailing partitionKeys.size() components of the partition
+ // directory. Never walk beyond them: the table location itself may contain foreign
+ // 'k=v' segments that must not leak into the registered spec. Registered values are
+ // RAW (unescaped); readers re-escape them when probing partition directories.
+ String[] components = new String[partitionKeys.size()];
+ Path current = partitionPath;
+ for (int i = partitionKeys.size() - 1; i >= 0; i--) {
+ if (current == null || current.getName().isEmpty()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition path '%s' has fewer than %s directory levels required "
+ + "by partition keys %s of table %s.",
+ partitionPath,
+ partitionKeys.size(),
+ partitionKeys,
+ tableIdentifier.getFullName()));
+ }
+ components[i] = current.getName();
+ current = current.getParent();
+ }
+
+ LinkedHashMap partitionSpec = new LinkedHashMap<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ String expectedKey = partitionKeys.get(i);
+ String component = components[i];
+ if (formatTablePartitionOnlyValueInPath) {
+ partitionSpec.put(expectedKey, PartitionPathUtils.unescapePathName(component));
+ } else {
+ int splitIndex = component.indexOf('=');
+ if (splitIndex < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition directory '%s' of partition path '%s' is not in "
+ + "'key=value' form expected for partition key '%s' "
+ + "of table %s.",
+ component,
+ partitionPath,
+ expectedKey,
+ tableIdentifier.getFullName()));
+ }
+ String parsedKey =
+ PartitionPathUtils.unescapePathName(component.substring(0, splitIndex));
+ String parsedValue =
+ PartitionPathUtils.unescapePathName(component.substring(splitIndex + 1));
+ if (!expectedKey.equals(parsedKey)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition directory '%s' of partition path '%s' declares "
+ + "partition key '%s' but partition key '%s' was "
+ + "expected at position %s for table %s.",
+ component,
+ partitionPath,
+ parsedKey,
+ expectedKey,
+ i,
+ tableIdentifier.getFullName()));
+ }
+ partitionSpec.put(expectedKey, parsedValue);
+ }
}
+ return partitionSpec;
}
private static Path buildPartitionPath(
@@ -208,24 +319,29 @@ private static Path buildPartitionPath(
if (partitionSpec.isEmpty() || partitionKeys.isEmpty()) {
throw new IllegalArgumentException("partitionSpec or partitionKeys is empty.");
}
- StringJoiner joiner = new StringJoiner("/");
+ LinkedHashMap orderedSpec = new LinkedHashMap<>();
for (int i = 0; i < partitionSpec.size(); i++) {
String key = partitionKeys.get(i);
if (partitionSpec.containsKey(key)) {
- if (formatTablePartitionOnlyValueInPath) {
- joiner.add(partitionSpec.get(key));
- } else {
- joiner.add(key + "=" + partitionSpec.get(key));
- }
+ orderedSpec.put(key, partitionSpec.get(key));
} else {
throw new RuntimeException("partitionSpec does not contain key: " + key);
}
}
- return new Path(location, joiner.toString());
+ return new Path(
+ location,
+ PartitionPathUtils.generatePartitionPathUtil(
+ orderedSpec, formatTablePartitionOnlyValueInPath));
}
@Override
public void abort(List commitMessages) {
+ // Once partition registration has started the data files are already committed and must be
+ // preserved: the catalog outcome is ambiguous (see partitionRegistrationFailure), so
+ // discarding the files here could delete data the catalog already references.
+ if (partitionRegistrationStarted) {
+ return;
+ }
try {
for (CommitMessage commitMessage : commitMessages) {
if (commitMessage instanceof TwoPhaseCommitMessage) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
index cb0edefc82fc..62fc6b09e3c5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTableScan.java
@@ -59,6 +59,7 @@
import javax.annotation.Nullable;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
@@ -81,9 +82,9 @@ public class FormatTableScan implements InnerTableScan {
private static final Logger LOG = LoggerFactory.getLogger(FormatTableScan.class);
- private final FormatTable table;
- private final CoreOptions coreOptions;
- @Nullable private PartitionPredicate partitionFilter;
+ protected final FormatTable table;
+ protected final CoreOptions coreOptions;
+ @Nullable protected PartitionPredicate partitionFilter;
@Nullable private final Integer limit;
private final long targetSplitSize;
private final long openFileCost;
@@ -142,7 +143,7 @@ public static boolean isDataFileName(String fileName) {
return fileName != null && !fileName.startsWith(".") && !fileName.startsWith("_");
}
- private BinaryRow toPartitionRow(LinkedHashMap partitionSpec) {
+ protected BinaryRow toPartitionRow(LinkedHashMap partitionSpec) {
RowType partitionType = table.partitionType();
GenericRow row =
convertSpecToInternalRow(partitionSpec, partitionType, table.defaultPartName());
@@ -160,7 +161,11 @@ public List splits() {
LinkedHashMap partitionSpec = pair.getKey();
BinaryRow partitionRow = toPartitionRow(partitionSpec);
if (partitionFilter == null || partitionFilter.test(partitionRow)) {
- splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow));
+ try {
+ splits.addAll(createSplits(fileIO, pair.getValue(), partitionRow));
+ } catch (FileNotFoundException e) {
+ onPartitionFileNotFound(partitionSpec, pair.getValue(), e);
+ }
}
}
} else {
@@ -177,7 +182,7 @@ public List splits() {
}
}
- List, Path>> findPartitions() {
+ protected List, Path>> findPartitions() {
LOG.debug(
"Find partitions for format table {}, partition filter: {}",
table.name(),
@@ -222,6 +227,14 @@ List, Path>> findPartitions() {
}
}
+ protected void onPartitionFileNotFound(
+ LinkedHashMap partitionSpec,
+ Path partitionPath,
+ FileNotFoundException exception)
+ throws IOException {
+ throw exception;
+ }
+
protected static List, Path>> generatePartitions(
List partitionKeys,
RowType partitionType,
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java
new file mode 100644
index 000000000000..7e16dee0839e
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/table/format/ManagedFormatTableScan.java
@@ -0,0 +1,209 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/** A format table scan whose partition visibility is owned by a catalog. */
+public class ManagedFormatTableScan extends FormatTableScan {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ManagedFormatTableScan.class);
+
+ private final FormatTableCatalogProvider catalogProvider;
+
+ public ManagedFormatTableScan(
+ FormatTable table,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable Integer limit) {
+ super(table, partitionFilter, limit);
+ FormatTableCatalogProvider provider = table.catalogProvider();
+ if (provider == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Managed format table %s has no catalog partition provider.",
+ table.fullName()));
+ }
+ this.catalogProvider = provider;
+ }
+
+ @Override
+ protected List, Path>> findPartitions() {
+ String partitionNamePattern = createPartitionNamePattern();
+ List partitions = catalogProvider.listPartitions(partitionNamePattern);
+ if (partitions.isEmpty() && partitionNamePattern == null) {
+ warnIfFilesystemPartitionsExist();
+ }
+ List, Path>> result = new ArrayList<>(partitions.size());
+ Path tablePath = new Path(table.location());
+ // Do not trust the catalog to be duplicate-free: a repeated spec would double every split
+ // of that partition and silently duplicate query results.
+ Set seenPartitionPaths = new HashSet<>(partitions.size());
+ for (Partition partition : partitions) {
+ LinkedHashMap partitionSpec = normalizeSpec(partition.spec());
+ String partitionPath =
+ PartitionPathUtils.generatePartitionPathUtil(
+ partitionSpec, coreOptions.formatTablePartitionOnlyValueInPath());
+ if (!seenPartitionPaths.add(partitionPath)) {
+ continue;
+ }
+ result.add(Pair.of(partitionSpec, new Path(tablePath, partitionPath)));
+ }
+ return result;
+ }
+
+ @Override
+ public List listPartitionEntries() {
+ List partitions = catalogProvider.listPartitions(null);
+ if (partitions.isEmpty()) {
+ warnIfFilesystemPartitionsExist();
+ }
+ List entries = new ArrayList<>(partitions.size());
+ for (Partition partition : partitions) {
+ entries.add(
+ new PartitionEntry(
+ toPartitionRow(normalizeSpec(partition.spec())),
+ partition.recordCount(),
+ partition.fileSizeInBytes(),
+ partition.fileCount(),
+ partition.lastFileCreationTime(),
+ partition.totalBuckets()));
+ }
+ return entries;
+ }
+
+ /**
+ * Warn when the catalog knows no partitions but the table directory contains subdirectories:
+ * typically a table that predates managed partition support (or was written by a client that
+ * does not register partitions) and needs a metadata sync before its data becomes visible.
+ */
+ private void warnIfFilesystemPartitionsExist() {
+ try {
+ for (FileStatus status : table.fileIO().listStatus(new Path(table.location()))) {
+ if (status.isDir() && !status.getPath().getName().startsWith(".")) {
+ LOG.warn(
+ "Managed format table {} has no partitions registered in the catalog "
+ + "but its location {} contains directories. Data written "
+ + "before enabling catalog-managed partitions (or by clients "
+ + "that do not register partitions) is invisible until the "
+ + "partition metadata is synced, e.g. with the Spark procedure "
+ + "sys.sync_format_table_metadata or MSCK REPAIR TABLE.",
+ table.fullName(),
+ table.location());
+ return;
+ }
+ }
+ } catch (IOException ignored) {
+ // Best-effort hint only; never fail or slow down the scan because of it.
+ }
+ }
+
+ /**
+ * Build the partition-name prefix pattern pushed down to the catalog, or {@code null} to list
+ * all partitions without pushdown. See {@link
+ * PartitionPathUtils#buildPartitionNamePrefixPattern} for the pattern contract; this returns
+ * {@code null} when there is no partition predicate, when the predicate has no leading equality
+ * prefix, or when the prefix cannot be expressed in the pattern contract.
+ */
+ @Nullable
+ String createPartitionNamePattern() {
+ Optional predicate = extractPartitionPredicate(partitionFilter);
+ if (!predicate.isPresent()) {
+ return null;
+ }
+
+ Map equalityPrefix =
+ extractLeadingEqualityPartitionSpecWhenOnlyAnd(
+ table.partitionKeys(), predicate.get(), table.partitionType());
+ return PartitionPathUtils.buildPartitionNamePrefixPattern(
+ table.partitionKeys(), equalityPrefix);
+ }
+
+ private LinkedHashMap normalizeSpec(Map spec) {
+ if (spec == null
+ || spec.size() != table.partitionKeys().size()
+ || !spec.keySet().containsAll(table.partitionKeys())) {
+ throw corruptPartitionSpec(spec);
+ }
+ LinkedHashMap normalized = new LinkedHashMap<>();
+ boolean onlyValueInPath = coreOptions.formatTablePartitionOnlyValueInPath();
+ for (String partitionKey : table.partitionKeys()) {
+ String value = spec.get(partitionKey);
+ // Catalog metadata is not trusted for path construction. In a value-only layout,
+ // reject complete path components such as '.' and '..' that would escape the table.
+ try {
+ PartitionPathUtils.validatePartitionValueForPath(value, onlyValueInPath);
+ } catch (IllegalArgumentException e) {
+ throw corruptPartitionSpec(spec);
+ }
+ normalized.put(partitionKey, value);
+ }
+ return normalized;
+ }
+
+ private IllegalStateException corruptPartitionSpec(@Nullable Map spec) {
+ return new IllegalStateException(
+ String.format(
+ "Catalog returned corrupt partition metadata %s for managed format table %s; "
+ + "expected exactly the partition keys %s with values usable as "
+ + "path components.",
+ spec, table.fullName(), table.partitionKeys()));
+ }
+
+ @Override
+ protected void onPartitionFileNotFound(
+ LinkedHashMap partitionSpec,
+ Path partitionPath,
+ FileNotFoundException exception) {
+ // A registered partition without a directory reads as empty, matching Hive semantics
+ // (e.g. ADD PARTITION before the first INSERT). Warn so genuine drift — a directory
+ // removed behind the catalog's back — is still discoverable.
+ LOG.warn(
+ "Partition '{}' of managed format table {} is registered in the catalog but its "
+ + "directory '{}' does not exist; treating the partition as empty. If the "
+ + "directory was removed on purpose, drop the partition or repair the "
+ + "metadata, e.g. with the Spark procedure sys.sync_format_table_metadata "
+ + "or MSCK REPAIR TABLE.",
+ PartitionPathUtils.generatePartitionName(partitionSpec, false),
+ table.fullName(),
+ partitionPath);
+ }
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
index fb7baf5b06b2..b17168cb3af2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/utils/PartitionPathUtils.java
@@ -109,13 +109,105 @@ public static String generatePartitionPathUtil(
suffixBuf.append(escapePathName(e.getKey()));
suffixBuf.append('=');
}
- suffixBuf.append(escapePathName(e.getValue()));
+ String value = e.getValue();
+ validatePartitionValueForPath(value, onlyValue);
+ suffixBuf.append(escapePathName(value));
i++;
}
suffixBuf.append(Path.SEPARATOR);
return suffixBuf.toString();
}
+ /**
+ * Generate a partition path without the trailing separator, e.g. {@code dt=20250101/hr=01}.
+ * This is the canonical partition name used when talking to a partition-managing catalog.
+ */
+ public static String generatePartitionName(
+ LinkedHashMap partitionSpec, boolean onlyValue) {
+ String path = generatePartitionPathUtil(partitionSpec, onlyValue);
+ return path.endsWith(Path.SEPARATOR)
+ ? path.substring(0, path.length() - Path.SEPARATOR.length())
+ : path;
+ }
+
+ /**
+ * Validate that a partition value is safe for the configured path layout. In a key-value
+ * layout, values such as {@code "."} are part of a component such as {@code "pt=."} and are
+ * safe. In a value-only layout, {@code "."} and {@code ".."} are complete path components and
+ * would resolve to a different directory.
+ */
+ public static void validatePartitionValueForPath(String value, boolean onlyValueInPath) {
+ if (value == null
+ || value.isEmpty()
+ || (onlyValueInPath && (".".equals(value) || "..".equals(value)))) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Partition value '%s' cannot be used as a partition path component.",
+ value));
+ }
+ }
+
+ /** Conservatively validate a value when the physical partition layout is unknown. */
+ public static void validatePartitionValueForPath(String value) {
+ validatePartitionValueForPath(value, true);
+ }
+
+ /** Validate every value of a partition spec for the configured path layout. */
+ public static void validatePartitionSpecForPath(
+ Map partitionSpec, boolean onlyValueInPath) {
+ for (String value : partitionSpec.values()) {
+ validatePartitionValueForPath(value, onlyValueInPath);
+ }
+ }
+
+ /** Conservatively validate a spec when the physical partition layout is unknown. */
+ public static void validatePartitionSpecForPath(Map partitionSpec) {
+ validatePartitionSpecForPath(partitionSpec, true);
+ }
+
+ /**
+ * Build the partition-name prefix pattern pushed down to a partition-managing catalog from the
+ * leading equality prefix of a partition predicate.
+ *
+ * Pattern contract (shared by every engine talking to the catalog): partition names are the
+ * escaped {@code key=value} form joined by {@code '/'}; {@code '%'} is the only wildcard and
+ * there is no escape sequence for it ({@code '_'} stays a literal). A complete spec matches the
+ * exact partition name; an incomplete prefix is suffixed with {@code '%'}.
+ *
+ *
Returns {@code null} whenever pushdown must be skipped and the caller should list all
+ * partitions instead: the equality prefix is empty, a prefix value is blank, or the escaped
+ * prefix contains a literal {@code '%'} that the contract cannot express.
+ */
+ @Nullable
+ public static String buildPartitionNamePrefixPattern(
+ List partitionKeys, Map equalityPrefix) {
+ if (equalityPrefix.isEmpty()) {
+ return null;
+ }
+ LinkedHashMap orderedPrefix = new LinkedHashMap<>();
+ for (String partitionKey : partitionKeys) {
+ if (!equalityPrefix.containsKey(partitionKey)) {
+ break;
+ }
+ String value = equalityPrefix.get(partitionKey);
+ if (StringUtils.isNullOrWhitespaceOnly(value)) {
+ return null;
+ }
+ orderedPrefix.put(partitionKey, value);
+ }
+ if (orderedPrefix.isEmpty()) {
+ return null;
+ }
+ String escapedPrefix = generatePartitionPath(orderedPrefix);
+ if (escapedPrefix.indexOf('%') >= 0) {
+ return null;
+ }
+ if (orderedPrefix.size() == partitionKeys.size()) {
+ return escapedPrefix.substring(0, escapedPrefix.length() - 1);
+ }
+ return escapedPrefix + '%';
+ }
+
public static List generatePartitionPaths(
List> partitions, RowType partitionType) {
return partitions.stream()
@@ -265,12 +357,45 @@ public static LinkedHashMap extractPartitionSpecFromPath(Path cu
return fullPartSpec;
}
+ /** Extract exactly the trailing key-value components for the declared partition keys. */
+ @Nullable
+ static LinkedHashMap extractPartitionSpecFromPath(
+ Path currPath, List partitionKeys) {
+ String[] values = new String[partitionKeys.size()];
+ Path current = currPath;
+ for (int i = partitionKeys.size() - 1; i >= 0; i--) {
+ if (current == null) {
+ return null;
+ }
+ Matcher matcher = PARTITION_NAME_PATTERN.matcher(current.getName());
+ if (!matcher.matches()
+ || !partitionKeys.get(i).equals(unescapePathName(matcher.group(1)))) {
+ return null;
+ }
+ values[i] = unescapePathName(matcher.group(2));
+ current = current.getParent();
+ }
+
+ LinkedHashMap spec = new LinkedHashMap<>();
+ for (int i = 0; i < partitionKeys.size(); i++) {
+ spec.put(partitionKeys.get(i), values[i]);
+ }
+ return spec;
+ }
+
public static LinkedHashMap extractPartitionSpecFromPathOnlyValue(
Path currPath, List partitionKeys) {
LinkedHashMap fullPartSpec = new LinkedHashMap<>();
String[] split = currPath.toString().split(Path.SEPARATOR);
for (int i = 0; i < partitionKeys.size(); i++) {
- fullPartSpec.put(partitionKeys.get(i), split[split.length - partitionKeys.size() + i]);
+ // Unescape the directory component so the extracted value is the RAW partition value,
+ // consistent with the key=value branch (extractPartitionSpecFromPath) and with the
+ // values the write path registers into a partition-managing catalog. Without this,
+ // directories containing escaped characters (e.g. a%3Ab) would round-trip to a
+ // different value than the one registered (a:b).
+ fullPartSpec.put(
+ partitionKeys.get(i),
+ unescapePathName(split[split.length - partitionKeys.size() + i]));
}
return fullPartSpec;
}
@@ -330,8 +455,9 @@ public static List, Path>> searchPartSpecAndP
part.getPath(), partitionKeys),
part.getPath()));
} else {
- LinkedHashMap spec = extractPartitionSpecFromPath(part.getPath());
- if (spec.size() != partitionKeys.size()) {
+ LinkedHashMap spec =
+ extractPartitionSpecFromPath(part.getPath(), partitionKeys);
+ if (spec == null) {
// illegal path, for example: /path/to/table/tmp/unknown, path without "="
continue;
}
@@ -360,8 +486,17 @@ private static FileStatus[] getFileStatusRecurse(
GenericRow values =
partitionType == null ? null : new GenericRow(partitionType.getFieldCount());
+ FileStatus fileStatus;
+ try {
+ fileStatus = fileIO.getFileStatus(path);
+ } catch (FileNotFoundException e) {
+ // A missing root simply means the table has no partitions yet.
+ return new FileStatus[0];
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to list files in " + path, e);
+ }
+
try {
- FileStatus fileStatus = fileIO.getFileStatus(path);
// Skip partition levels already fixed by the scan-path prefix.
int levelOffset = partitionKeys.size() - expectLevel;
listStatusRecursively(
@@ -378,9 +513,10 @@ private static FileStatus[] getFileStatusRecurse(
defaultPartValue,
levelOffset,
values);
- } catch (FileNotFoundException e) {
- return new FileStatus[0];
} catch (IOException e) {
+ // Never degrade a mid-scan failure into an empty listing: callers diff this result
+ // against partition metadata and an incomplete listing would deregister partitions
+ // that still exist.
throw new RuntimeException("Failed to list files in " + path, e);
}
@@ -412,7 +548,15 @@ private static void listStatusRecursively(
}
if (fileStatus.isDir()) {
- for (FileStatus stat : fileIO.listStatus(fileStatus.getPath())) {
+ FileStatus[] children;
+ try {
+ children = fileIO.listStatus(fileStatus.getPath());
+ } catch (FileNotFoundException e) {
+ // The directory vanished after the parent listed it: the partitions beneath it
+ // are gone, skipping just this subtree keeps the rest of the listing complete.
+ return;
+ }
+ for (FileStatus stat : children) {
int partitionKeyIndex = levelOffset + level;
String partitionKey = partitionKeys.get(partitionKeyIndex);
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
index 0961050f110d..3e91251ae812 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CachingCatalogTest.java
@@ -368,6 +368,32 @@ public void testCreatePartitionsWithIgnoreIfExistsInvalidatesPartitionCache() th
assertThat(catalog.listPartitions(identifier)).containsExactly(created);
}
+ @Test
+ public void testCreatePartitionsFailureInvalidatesPartitionCache() throws Exception {
+ Catalog wrapped = Mockito.mock(Catalog.class);
+ TestableCachingCatalog catalog =
+ new TestableCachingCatalog(wrapped, EXPIRATION_TTL, ticker);
+ Identifier identifier = new Identifier("db", "tbl");
+ Map spec = singletonMap("dt", "20260717");
+ Partition created = new Partition(spec, 0, 0, 0, 0, -1, false);
+ List stored = new ArrayList<>();
+ when(wrapped.listPartitions(identifier)).thenAnswer(ignored -> new ArrayList<>(stored));
+ RuntimeException responseLost = new RuntimeException("response lost");
+ Mockito.doAnswer(
+ ignored -> {
+ stored.add(created);
+ throw responseLost;
+ })
+ .when(wrapped)
+ .createPartitions(identifier, singletonList(spec));
+
+ assertThat(catalog.listPartitions(identifier)).isEmpty();
+ assertThatThrownBy(() -> catalog.createPartitions(identifier, singletonList(spec)))
+ .isSameAs(responseLost);
+
+ assertThat(catalog.listPartitions(identifier)).containsExactly(created);
+ }
+
@Test
public void testDeadlock() throws Exception {
Catalog underlyCatalog = this.catalog;
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
new file mode 100644
index 000000000000..edcc79af157c
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogUtilsTest.java
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+
+import static org.apache.paimon.CoreOptions.FILE_FORMAT;
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_IMPLEMENTATION;
+import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
+import static org.apache.paimon.CoreOptions.PATH;
+import static org.apache.paimon.CoreOptions.TYPE;
+import static org.apache.paimon.TableType.FORMAT_TABLE;
+import static org.apache.paimon.catalog.CatalogUtils.loadTable;
+import static org.apache.paimon.catalog.CatalogUtils.validateCreateTable;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link CatalogUtils}. */
+class CatalogUtilsTest {
+
+ @Test
+ void testRejectEngineImplementationForManagedFormatTableOnCreate() {
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .option(TYPE.key(), FORMAT_TABLE.toString())
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(FORMAT_TABLE_IMPLEMENTATION.key(), "engine")
+ .build();
+
+ assertThatThrownBy(() -> validateCreateTable(schema, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key())
+ .hasMessageContaining(FORMAT_TABLE_IMPLEMENTATION.key());
+ }
+
+ @Test
+ void testEngineImplementationManagedFormatTableLoadsAsUnmanaged() throws Exception {
+ // The option combination is invalid for managed partitions, but an existing table must
+ // stay loadable: it degrades to an unmanaged format table instead of failing.
+ Table table = loadManagedFormatTable("engine", true);
+ assertUnmanagedFormatTable(table);
+ }
+
+ @Test
+ void testManagedFormatTableOutsideCapableCatalogLoadsAsUnmanaged() throws Exception {
+ // A catalog that cannot manage partitions (e.g. Hive) must not reject existing tables that
+ // carry the (previously inert) option; it loads them as unmanaged format tables.
+ Table table = loadManagedFormatTable("paimon", false);
+ assertUnmanagedFormatTable(table);
+ }
+
+ @Test
+ void testLoadManagedFormatTableFromCapableCatalog() throws Exception {
+ Table table = loadManagedFormatTable("paimon", true);
+ assertThat(table).isInstanceOf(FormatTable.class);
+ FormatTable formatTable = (FormatTable) table;
+ assertThat(
+ new org.apache.paimon.CoreOptions(formatTable.options())
+ .partitionedTableInMetastore())
+ .isTrue();
+ assertThat(formatTable.catalogProvider()).isNotNull();
+ }
+
+ private static void assertUnmanagedFormatTable(Table table) {
+ assertThat(table).isInstanceOf(FormatTable.class);
+ FormatTable formatTable = (FormatTable) table;
+ assertThat(
+ new org.apache.paimon.CoreOptions(formatTable.options())
+ .partitionedTableInMetastore())
+ .isFalse();
+ assertThat(formatTable.catalogProvider()).isNull();
+ }
+
+ private static Table loadManagedFormatTable(
+ String formatTableImplementation, boolean supportsManagedPartitions) throws Exception {
+ Identifier identifier = Identifier.create("managed_db", "managed_table");
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .option(TYPE.key(), FORMAT_TABLE.toString())
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(FORMAT_TABLE_IMPLEMENTATION.key(), formatTableImplementation)
+ .option(FILE_FORMAT.key(), "parquet")
+ .option(PATH.key(), "file:///tmp/managed_db.db/managed_table")
+ .build();
+ TableMetadata metadata =
+ new TableMetadata(TableSchema.create(0, schema), false, "managed-table-id");
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.catalogLoader()).thenReturn(() -> catalog);
+ when(catalog.supportsManagedFormatTablePartitions()).thenReturn(supportsManagedPartitions);
+
+ return loadTable(
+ catalog,
+ identifier,
+ path -> new LocalFileIO(),
+ path -> new LocalFileIO(),
+ ignored -> metadata,
+ null,
+ null,
+ null,
+ supportsManagedPartitions);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
index 04f67c018d1b..7bb122e1fea6 100644
--- a/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/catalog/FileSystemCatalogTest.java
@@ -18,9 +18,15 @@
package org.apache.paimon.catalog;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.TableType;
import org.apache.paimon.fs.Path;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.schema.SchemaChange;
+import org.apache.paimon.schema.SchemaManager;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.Table;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
@@ -28,6 +34,7 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link FileSystemCatalog}. */
@@ -77,4 +84,49 @@ public void testAlterDatabase() throws Exception {
false))
.isInstanceOf(UnsupportedOperationException.class);
}
+
+ @Test
+ public void testResetManagedPartitionsOnStrandedFormatTable() throws Exception {
+ String database = "stranded_managed_format_table_db";
+ Identifier identifier = Identifier.create(database, "stranded_format_table");
+ catalog.createDatabase(database, false);
+ // Write the schema file directly to simulate a format table stranded with the
+ // REST-only option in a filesystem catalog.
+ Schema schema =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true")
+ .build();
+ Path tablePath =
+ ((FileSystemCatalog) DelegateCatalog.rootCatalog(catalog))
+ .getTableLocation(identifier);
+ new SchemaManager(fileIO, tablePath).createTable(schema);
+ // A catalog that cannot manage partitions must still load such a stranded table: the
+ // option is ignored and the table degrades to an unmanaged format table (readable), rather
+ // than becoming permanently unreadable.
+ Table stranded = catalog.getTable(identifier);
+ assertThat(stranded).isInstanceOf(FormatTable.class);
+ assertThat(new CoreOptions(stranded.options()).partitionedTableInMetastore()).isFalse();
+
+ // Alters that do not touch managed-sensitive options skip the managed validation.
+ catalog.alterTable(
+ identifier,
+ Lists.newArrayList(SchemaChange.setOption("custom.unrelated-option", "value")),
+ false);
+
+ // The documented remediation must succeed on a non-REST catalog.
+ catalog.alterTable(
+ identifier,
+ Lists.newArrayList(
+ SchemaChange.removeOption(CoreOptions.METASTORE_PARTITIONED_TABLE.key())),
+ false);
+
+ assertThat(catalog.getTable(identifier).options())
+ .doesNotContainKey(CoreOptions.METASTORE_PARTITIONED_TABLE.key())
+ .containsEntry("custom.unrelated-option", "value");
+ }
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
index b4af3194b72f..c05f45875ed4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/MockRESTCatalogTest.java
@@ -26,6 +26,8 @@
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.catalog.TableQueryAuthResult;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.CatalogOptions;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.FieldRef;
@@ -46,6 +48,8 @@
import org.apache.paimon.rest.exceptions.NotImplementedException;
import org.apache.paimon.rest.responses.ConfigResponse;
import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.format.FormatTableCatalogProvider;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.JsonSerdeUtil;
@@ -247,6 +251,96 @@ void testManagedFormatTablePartitionListingDoesNotFallback() throws Exception {
.isInstanceOf(NotImplementedException.class);
}
+ @Test
+ void testCreatePartitionFailureInvalidatesManagedListingCache() throws Exception {
+ Identifier identifier = createManagedFormatTable();
+ FormatTable table = (FormatTable) restCatalog.getTable(identifier);
+ FormatTableCatalogProvider provider = table.catalogProvider();
+ assertThat(provider).isNotNull();
+ assertThat(provider.listPartitions(null)).isEmpty();
+ Map partition = Collections.singletonMap("dt", "20260717");
+ restCatalogServer.failNextCreatePartitionsResponse();
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.createPartitions(
+ identifier, Collections.singletonList(partition)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Injected response failure");
+
+ assertThat(provider.listPartitions(null))
+ .extracting(org.apache.paimon.partition.Partition::spec)
+ .containsExactly(partition);
+ }
+
+ @Test
+ void testDropPartitionFailureInvalidatesManagedListingCache() throws Exception {
+ Identifier identifier = createManagedFormatTable();
+ FormatTable table = (FormatTable) restCatalog.getTable(identifier);
+ FormatTableCatalogProvider provider = table.catalogProvider();
+ assertThat(provider).isNotNull();
+ Map partition = Collections.singletonMap("dt", "20260717");
+ restCatalog.createPartitions(identifier, Collections.singletonList(partition));
+ assertThat(provider.listPartitions(null))
+ .extracting(org.apache.paimon.partition.Partition::spec)
+ .containsExactly(partition);
+ restCatalogServer.failNextDropPartitionsResponse();
+
+ assertThatThrownBy(
+ () ->
+ restCatalog.dropPartitions(
+ identifier, Collections.singletonList(partition)))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Injected response failure");
+
+ assertThat(provider.listPartitions(null)).isEmpty();
+ }
+
+ @Test
+ void testRejectExternalManagedFormatTableBeforeCreate() throws Exception {
+ Identifier identifier = Identifier.create("db1", "external_managed_format_table");
+ restCatalog.createDatabase(identifier.getDatabaseName(), true);
+ String externalPath = dataPath + "/external-managed-format-table";
+ Schema schema =
+ Schema.newBuilder()
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(CoreOptions.FILE_FORMAT.key(), "parquet")
+ .option(CoreOptions.PATH.key(), externalPath)
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .build();
+
+ assertThatThrownBy(() -> restCatalog.createTable(identifier, schema, false))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("internal table");
+ assertThat(restCatalog.listTables(identifier.getDatabaseName()))
+ .doesNotContain(identifier.getTableName());
+ assertThat(LocalFileIO.create().exists(new Path(externalPath))).isFalse();
+ }
+
+ @Test
+ void testRoundTrippedManagedFormatTableReplacePassesClientValidation() throws Exception {
+ Identifier identifier = createManagedFormatTable();
+ FormatTable existing = (FormatTable) restCatalog.getTable(identifier);
+ Schema replacement =
+ Schema.newBuilder()
+ .column("id", DataTypes.INT())
+ .column("dt", DataTypes.STRING())
+ .partitionKeys("dt")
+ .options(existing.options())
+ .build();
+ assertThat(replacement.options())
+ .containsEntry(CoreOptions.PATH.key(), existing.location());
+
+ // The mock service does not implement Format Table replacement. Reaching that response
+ // proves the REST client accepted the unchanged synthetic path from the loaded table.
+ assertThatThrownBy(() -> restCatalog.replaceTable(identifier, replacement, false))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessageContaining("replaceTable does not support format tables");
+ }
+
private Identifier createManagedFormatTable() throws Exception {
Identifier identifier = Identifier.create("db1", "managed_partition_table");
restCatalog.createDatabase(identifier.getDatabaseName(), true);
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
index 3a0e1539eea1..385b27cb9fbf 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -212,6 +212,9 @@ public class RESTCatalogServer {
private final List> receivedHeaders = new ArrayList<>();
private volatile boolean partitionListingSupported = true;
+ private volatile boolean failNextCreatePartitionsResponse;
+ private volatile boolean failNextDropPartitionsResponse;
+ private volatile int tableGetCount;
public RESTCatalogServer(
String dataPath, AuthProvider authProvider, ConfigResponse config, String warehouse) {
@@ -279,6 +282,22 @@ public void setPartitionListingSupported(boolean partitionListingSupported) {
this.partitionListingSupported = partitionListingSupported;
}
+ public void failNextCreatePartitionsResponse() {
+ this.failNextCreatePartitionsResponse = true;
+ }
+
+ public void failNextDropPartitionsResponse() {
+ this.failNextDropPartitionsResponse = true;
+ }
+
+ public int tableGetCount() {
+ return tableGetCount;
+ }
+
+ public void resetTableGetCount() {
+ tableGetCount = 0;
+ }
+
public void addNoPermissionDatabase(String database) {
noPermissionDatabases.add(database);
}
@@ -1733,6 +1752,7 @@ private MockResponse tableHandle(String method, String data, Identifier identifi
}
switch (method) {
case "GET":
+ tableGetCount++;
TableMetadata tableMetadata;
if (identifier.isSystemTable()) {
TableSchema schema = catalog.loadTableSchema(identifier);
@@ -1897,6 +1917,16 @@ private MockResponse partitionsApiHandle(
existed.add(spec);
}
}
+ if (failNextCreatePartitionsResponse) {
+ failNextCreatePartitionsResponse = false;
+ return mockResponse(
+ new ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_PARTITION,
+ tableIdentifier.getFullName(),
+ "Injected response failure after creating partitions.",
+ 400),
+ 400);
+ }
return mockResponse(new CreatePartitionsResponse(created, existed), 200);
default:
return new MockResponse().setResponseCode(404);
@@ -1940,6 +1970,16 @@ private MockResponse dropPartitionsHandle(String data, Identifier tableIdentifie
}
return false;
});
+ if (failNextDropPartitionsResponse) {
+ failNextDropPartitionsResponse = false;
+ return mockResponse(
+ new ErrorResponse(
+ ErrorResponse.RESOURCE_TYPE_PARTITION,
+ tableIdentifier.getFullName(),
+ "Injected response failure after dropping partitions.",
+ 400),
+ 400);
+ }
return mockResponse(new DropPartitionsResponse(dropped, missing), 200);
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 504dc6fab6ec..3f13c1cab844 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -76,6 +76,7 @@
import org.apache.paimon.schema.SchemaChange;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FormatTable;
import org.apache.paimon.table.Instant;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableSnapshot;
@@ -1639,6 +1640,63 @@ void testDropPartitionsLoadsNonManagedTableOnce() throws Exception {
Mockito.verify(catalogSpy, Mockito.times(1)).getTable(identifier);
}
+ @Test
+ void testManagedFormatTableCommitAndScanMatchesFileSystemMode() throws Exception {
+ Identifier identifier = Identifier.create("format_partition_db", "managed_scan_table");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+ catalog.createTable(
+ identifier,
+ Schema.newBuilder()
+ .option(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString())
+ .option(METASTORE_PARTITIONED_TABLE.key(), "true")
+ .option(CoreOptions.FILE_FORMAT.key(), "csv")
+ .column("id", DataTypes.INT())
+ .column("year", DataTypes.INT())
+ .column("month", DataTypes.INT())
+ .partitionKeys("year", "month")
+ .build(),
+ false);
+
+ FormatTable managedTable = (FormatTable) catalog.getTable(identifier);
+ BatchWriteBuilder writeBuilder = managedTable.newBatchWriteBuilder();
+ try (BatchTableWrite write = writeBuilder.newWrite();
+ BatchTableCommit commit = writeBuilder.newCommit()) {
+ write.write(GenericRow.of(1, 2024, 10));
+ write.write(GenericRow.of(2, 2025, 10));
+ write.write(GenericRow.of(3, 2025, 11));
+ commit.commit(write.prepareCommit());
+ }
+
+ List> registeredPartitions =
+ Arrays.asList(
+ ImmutableMap.of("year", "2024", "month", "10"),
+ ImmutableMap.of("year", "2025", "month", "10"),
+ ImmutableMap.of("year", "2025", "month", "11"));
+ assertThat(catalog.listPartitions(identifier).stream().map(Partition::spec))
+ .containsExactlyInAnyOrderElementsOf(registeredPartitions);
+ assertThat(managedTable.newReadBuilder().newScan().getClass().getSimpleName())
+ .isEqualTo("ManagedFormatTableScan");
+
+ Map fileSystemOptions = new HashMap<>(managedTable.options());
+ fileSystemOptions.put(METASTORE_PARTITIONED_TABLE.key(), "false");
+ FormatTable fileSystemTable =
+ FormatTable.builder()
+ .fileIO(managedTable.fileIO())
+ .identifier(Identifier.create("format_partition_db", "filesystem_scan"))
+ .rowType(managedTable.rowType())
+ .partitionKeys(managedTable.partitionKeys())
+ .location(managedTable.location())
+ .format(managedTable.format())
+ .options(fileSystemOptions)
+ .catalogContext(managedTable.catalogContext())
+ .build();
+
+ Map partitionFilter = singletonMap("year", "2025");
+ assertThat(read(managedTable, null, null, partitionFilter, null))
+ .containsExactlyInAnyOrderElementsOf(
+ read(fileSystemTable, null, null, partitionFilter, null));
+ }
+
@Test
void testListPartitions() throws Exception {
innerTestListPartitions(true);
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java
new file mode 100644
index 000000000000..29fe9f4fd7fe
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/FormatTableCompatibilityTest.java
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.format.FormatTableCatalogProvider;
+import org.apache.paimon.table.format.FormatTableCommit;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
+import static org.apache.paimon.CoreOptions.METASTORE_PARTITIONED_TABLE;
+import static org.apache.paimon.CoreOptions.READ_BATCH_SIZE;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Compatibility tests for the public {@link FormatTable} API. */
+class FormatTableCompatibilityTest {
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCopyRejectsManagedPartitionStateChange(boolean managed) {
+ FormatTable table = formatTable(Boolean.toString(managed));
+
+ assertThatThrownBy(
+ () ->
+ table.copy(
+ Collections.singletonMap(
+ METASTORE_PARTITIONED_TABLE.key(),
+ Boolean.toString(!managed))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key());
+ }
+
+ @Test
+ void testCopyAllowsSemanticallyEquivalentManagedPartitionOption() {
+ FormatTable table = formatTable("TRUE");
+
+ FormatTable copied =
+ table.copy(Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), "true"));
+
+ assertThat(copied).isNotSameAs(table);
+ assertThat(copied.options())
+ .containsEntry(METASTORE_PARTITIONED_TABLE.key(), Boolean.TRUE.toString());
+ }
+
+ @Test
+ void testCopyRejectsChangingAbsentManagedPartitionDefault() {
+ FormatTable table = formatTable(Collections.emptyMap(), null);
+
+ assertThat(table.options()).doesNotContainKey(METASTORE_PARTITIONED_TABLE.key());
+ assertThatThrownBy(
+ () ->
+ table.copy(
+ Collections.singletonMap(
+ METASTORE_PARTITIONED_TABLE.key(), "true")))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(METASTORE_PARTITIONED_TABLE.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testManagedCopyRejectsPhysicalPartitionLayoutChange(boolean onlyValueInPath) {
+ Map options = new LinkedHashMap<>();
+ options.put(METASTORE_PARTITIONED_TABLE.key(), "true");
+ options.put(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(), Boolean.toString(onlyValueInPath));
+ FormatTable table = formatTable(options, null);
+
+ assertThatThrownBy(
+ () ->
+ table.copy(
+ Collections.singletonMap(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(!onlyValueInPath))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testCopyPreservesManagedStateAndCatalogProviderForUnrelatedOption(boolean managed) {
+ Identifier identifier = Identifier.create("managed_db", "managed_table");
+ FormatTableCatalogProvider catalogProvider =
+ new FormatTableCatalogProvider(identifier, () -> null);
+ Map options = new LinkedHashMap<>();
+ options.put(METASTORE_PARTITIONED_TABLE.key(), Boolean.toString(managed));
+ FormatTable table = formatTable(options, catalogProvider);
+
+ FormatTable copied = table.copy(Collections.singletonMap(READ_BATCH_SIZE.key(), "128"));
+
+ assertThat(copied).isNotSameAs(table);
+ assertThat(copied.options())
+ .containsEntry(READ_BATCH_SIZE.key(), "128")
+ .containsEntry(METASTORE_PARTITIONED_TABLE.key(), Boolean.toString(managed));
+ assertThat(
+ new org.apache.paimon.CoreOptions(copied.options())
+ .partitionedTableInMetastore())
+ .isEqualTo(managed);
+ assertThat(copied.catalogProvider()).isSameAs(catalogProvider);
+ assertThat(table.options())
+ .containsExactlyEntriesOf(options)
+ .doesNotContainKey(READ_BATCH_SIZE.key());
+ }
+
+ @Test
+ void testManagedCatalogExtensionKeepsExistingImplementationsCompatible() throws Exception {
+ assertThat(FormatTable.class.getMethod("catalogProvider").isDefault()).isTrue();
+ assertThat(
+ FormatTable.FormatTableImpl.class.getConstructor(
+ FileIO.class,
+ Identifier.class,
+ RowType.class,
+ List.class,
+ String.class,
+ FormatTable.Format.class,
+ Map.class,
+ String.class,
+ CatalogContext.class))
+ .isNotNull();
+ assertThat(
+ FormatTableCommit.class.getConstructor(
+ String.class,
+ List.class,
+ FileIO.class,
+ boolean.class,
+ boolean.class,
+ Identifier.class,
+ Map.class,
+ String.class,
+ CatalogContext.class))
+ .isNotNull();
+ }
+
+ private static FormatTable formatTable(String managed) {
+ return formatTable(
+ Collections.singletonMap(METASTORE_PARTITIONED_TABLE.key(), managed), null);
+ }
+
+ private static FormatTable formatTable(
+ Map options, FormatTableCatalogProvider catalogProvider) {
+ return FormatTable.builder()
+ .fileIO(LocalFileIO.create())
+ .identifier(Identifier.create("managed_db", "managed_table"))
+ .rowType(
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("dt", DataTypes.STRING())
+ .build())
+ .partitionKeys(Collections.singletonList("dt"))
+ .location("file:///warehouse/managed_table")
+ .format(FormatTable.Format.PARQUET)
+ .options(options)
+ .catalogProvider(catalogProvider)
+ .build();
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
new file mode 100644
index 000000000000..2bc74ca99bd6
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTableCommitTest.java
@@ -0,0 +1,247 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.RenamingTwoPhaseOutputStream;
+import org.apache.paimon.fs.TwoPhaseOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/** Tests for {@link FormatTableCommit}. */
+class FormatTableCommitTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testPartitionRegistrationFailurePreservesCommittedFiles() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path targetPath = new Path(tablePath, "year=2025/month=10/data-1.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit();
+ FormatTableCatalogProvider catalogProvider = mock(FormatTableCatalogProvider.class);
+ RuntimeException registrationFailure =
+ new RuntimeException("Catalog partition registration unavailable");
+ doThrow(registrationFailure).when(catalogProvider).createPartitions(anyList());
+
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ false,
+ Identifier.create("managed_db", "managed_table"),
+ null,
+ null,
+ null,
+ catalogProvider);
+ CommitMessage message = new TwoPhaseCommitMessage(committer);
+
+ assertThatThrownBy(() -> commit.commit(Collections.singletonList(message)))
+ .isInstanceOf(RuntimeException.class)
+ .hasMessageContaining("Committed data files were preserved")
+ .hasMessageContaining("managed_db.managed_table")
+ .hasMessageContaining("MSCK REPAIR TABLE")
+ .hasRootCauseMessage("Catalog partition registration unavailable");
+
+ assertThat(fileIO.exists(targetPath)).isTrue();
+ verify(catalogProvider).createPartitions(anyList());
+
+ // Flink calls abort on the same commit object after a failed commit.
+ commit.abort(Collections.singletonList(message));
+ assertThat(fileIO.exists(targetPath)).isTrue();
+ }
+
+ @Test
+ void testFileCommitFailureStillDiscardsUncommittedFiles() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ TwoPhaseOutputStream.Committer committer = mock(TwoPhaseOutputStream.Committer.class);
+ doThrow(new IOException("data commit failed")).when(committer).commit(fileIO);
+ FormatTableCatalogProvider catalogProvider = mock(FormatTableCatalogProvider.class);
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ false,
+ false,
+ Identifier.create("managed_db", "managed_table"),
+ null,
+ null,
+ null,
+ catalogProvider);
+ CommitMessage message = new TwoPhaseCommitMessage(committer);
+
+ assertThatThrownBy(() -> commit.commit(Collections.singletonList(message)))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseMessage("data commit failed");
+
+ verify(committer).discard(fileIO);
+ verify(catalogProvider, never()).createPartitions(anyList());
+ }
+
+ @Test
+ void testRegistersRawPartitionValuesForEscapedPath() throws Exception {
+ Path tablePath = new Path(tempDir.toUri());
+ LinkedHashMap rawSpec = new LinkedHashMap<>();
+ rawSpec.put("year", "2025");
+ rawSpec.put("month", "a b:c");
+ // The writer escapes partition values when building the directory layout.
+ String partitionDir = PartitionPathUtils.generatePartitionPathUtil(rawSpec, false);
+ assertThat(partitionDir).isEqualTo("year=2025/month=a b%3Ac/");
+
+ FormatTableCatalogProvider catalogProvider =
+ commitPartitionedFile(tablePath, false, partitionDir);
+
+ // The catalog must receive RAW values; readers re-escape them when probing directories.
+ assertThat(registeredSpec(catalogProvider))
+ .containsExactly(entry("year", "2025"), entry("month", "a b:c"));
+ }
+
+ @Test
+ void testForeignKeyValueSegmentsInLocationDoNotLeakIntoSpec() throws Exception {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "env=prod/warehouse/tbl");
+
+ FormatTableCatalogProvider catalogProvider =
+ commitPartitionedFile(tablePath, false, "year=2025/month=10");
+
+ assertThat(registeredSpec(catalogProvider))
+ .containsExactly(entry("year", "2025"), entry("month", "10"));
+ }
+
+ @Test
+ void testValueOnlyPathUnderForeignKeyValueSegmentRegistersRawValues() throws Exception {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "env=prod/warehouse/tbl");
+ LinkedHashMap rawSpec = new LinkedHashMap<>();
+ rawSpec.put("year", "2025");
+ rawSpec.put("month", "a:b");
+ String partitionDir = PartitionPathUtils.generatePartitionPathUtil(rawSpec, true);
+ assertThat(partitionDir).isEqualTo("2025/a%3Ab/");
+
+ FormatTableCatalogProvider catalogProvider =
+ commitPartitionedFile(tablePath, true, partitionDir);
+
+ assertThat(registeredSpec(catalogProvider))
+ .containsExactly(entry("year", "2025"), entry("month", "a:b"));
+ }
+
+ @Test
+ void testMismatchedPartitionKeyInPathFailsWithClearMessage() {
+ Path tablePath = new Path(tempDir.toUri());
+
+ assertThatThrownBy(() -> commitPartitionedFile(tablePath, false, "year=2025/day=10"))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .rootCause()
+ .hasMessageContaining("declares partition key 'day'")
+ .hasMessageContaining("partition key 'month' was expected");
+ }
+
+ @Test
+ void testValueOnlyStaticPartitionCannotEscapeTableLocation() throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path parentPath = new Path(tempDir.toUri());
+ Path tablePath = new Path(parentPath, "table");
+ Path siblingPath = new Path(parentPath, "keep");
+ fileIO.mkdirs(tablePath);
+ fileIO.mkdirs(siblingPath);
+ Map staticPartition = Collections.singletonMap("year", "..");
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tablePath.toString(),
+ Collections.singletonList("year"),
+ fileIO,
+ true,
+ true,
+ Identifier.create("managed_db", "managed_table"),
+ staticPartition,
+ null,
+ null,
+ null);
+
+ assertThatThrownBy(() -> commit.commit(Collections.emptyList()))
+ .isInstanceOf(RuntimeException.class)
+ .hasRootCauseInstanceOf(IllegalArgumentException.class)
+ .hasRootCauseMessage(
+ "Partition value '..' cannot be used as a partition path component.");
+ assertThat(fileIO.exists(tablePath)).isTrue();
+ assertThat(fileIO.exists(siblingPath)).isTrue();
+ }
+
+ private FormatTableCatalogProvider commitPartitionedFile(
+ Path tableLocation, boolean onlyValueInPath, String partitionDir) throws Exception {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path targetPath = new Path(new Path(tableLocation, partitionDir), "data-1.csv");
+ RenamingTwoPhaseOutputStream outputStream =
+ new RenamingTwoPhaseOutputStream(fileIO, targetPath, false);
+ outputStream.write(1);
+ TwoPhaseOutputStream.Committer committer = outputStream.closeForCommit();
+ FormatTableCatalogProvider catalogProvider = mock(FormatTableCatalogProvider.class);
+ FormatTableCommit commit =
+ new FormatTableCommit(
+ tableLocation.toString(),
+ Arrays.asList("year", "month"),
+ fileIO,
+ onlyValueInPath,
+ false,
+ Identifier.create("managed_db", "managed_table"),
+ null,
+ null,
+ null,
+ catalogProvider);
+ commit.commit(Collections.singletonList(new TwoPhaseCommitMessage(committer)));
+ return catalogProvider;
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static Map registeredSpec(FormatTableCatalogProvider catalogProvider) {
+ ArgumentCaptor>> captor =
+ ArgumentCaptor.forClass((Class) List.class);
+ verify(catalogProvider).createPartitions(captor.capture());
+ assertThat(captor.getValue()).hasSize(1);
+ return captor.getValue().get(0);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java b/paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java
new file mode 100644
index 000000000000..0ca741386eed
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/table/format/ManagedFormatTableScanTest.java
@@ -0,0 +1,553 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.PagedList;
+import org.apache.paimon.catalog.Catalog;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.PartitionPathUtils;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for managed format table partition discovery. */
+class ManagedFormatTableScanTest {
+
+ private static final Identifier IDENTIFIER = Identifier.create("managed_db", "managed_table");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testProviderRegistersLargeBatchesInBoundedRequests() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ FormatTableCatalogProvider provider = provider(catalog);
+ List> specs = new ArrayList<>();
+ for (int index = 0; index < 2500; index++) {
+ specs.add(Collections.singletonMap("year", String.format("y%04d", index)));
+ }
+
+ provider.createPartitions(specs);
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ org.mockito.ArgumentCaptor>> batches =
+ (org.mockito.ArgumentCaptor) org.mockito.ArgumentCaptor.forClass(List.class);
+ verify(catalog, times(3)).createPartitions(eq(IDENTIFIER), batches.capture());
+ assertThat(batches.getAllValues().get(0)).hasSize(1000);
+ assertThat(batches.getAllValues().get(1)).hasSize(1000);
+ assertThat(batches.getAllValues().get(2)).hasSize(500);
+ assertThat(
+ batches.getAllValues().stream()
+ .flatMap(List::stream)
+ .collect(java.util.stream.Collectors.toList()))
+ .containsExactlyElementsOf(specs);
+ }
+
+ @Test
+ void testProviderPaginatesAndCachesMatchingPattern() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition october = partition("2025", "10");
+ Partition november = partition("2025", "11");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenReturn(new PagedList<>(Collections.singletonList(october), "next"));
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), eq("next"), eq("year=2025/%")))
+ .thenReturn(new PagedList<>(Collections.singletonList(november), null));
+
+ FormatTableCatalogProvider provider = provider(catalog);
+
+ assertThat(provider.listPartitions("year=2025/%")).containsExactly(october, november);
+ assertThat(provider.listPartitions("year=2025/%")).containsExactly(october, november);
+ verify(catalog, times(1)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%");
+ verify(catalog, times(1)).listPartitionsPaged(IDENTIFIER, 1000, "next", "year=2025/%");
+ }
+
+ @Test
+ void testProviderRejectsRepeatedPageTokenCycle() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), "a"));
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), eq("a"), isNull()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), "b"))
+ .thenThrow(new AssertionError("provider requested page token 'a' twice"));
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), eq("b"), isNull()))
+ .thenReturn(new PagedList<>(Collections.emptyList(), "a"));
+
+ assertThatThrownBy(() -> provider(catalog).listPartitions(null))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("repeated partition page token 'a'")
+ .hasMessageContaining(IDENTIFIER.getFullName());
+ }
+
+ @Test
+ void testCreatePartitionsInvalidatesCachedPattern() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition oldPartition = partition("2025", "10");
+ Partition newPartition = partition("2025", "11");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(oldPartition), null),
+ new PagedList<>(Arrays.asList(oldPartition, newPartition), null));
+
+ FormatTableCatalogProvider provider = provider(catalog);
+ assertThat(provider.listPartitions("year=2025/%")).containsExactly(oldPartition);
+ assertThat(provider.listPartitions("year=2025/%")).containsExactly(oldPartition);
+
+ List> created = Collections.singletonList(newPartition.spec());
+ provider.createPartitions(created);
+
+ assertThat(provider.listPartitions("year=2025/%"))
+ .containsExactly(oldPartition, newPartition);
+ verify(catalog).createPartitions(IDENTIFIER, created);
+ verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%");
+ }
+
+ @Test
+ void testCreatePartitionsInvalidatesOtherProviderInSameProcess() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition oldPartition = partition("2025", "10");
+ Partition newPartition = partition("2025", "11");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(oldPartition), null),
+ new PagedList<>(Arrays.asList(oldPartition, newPartition), null));
+ FormatTableCatalogProvider readerProvider = provider(catalog);
+ FormatTableCatalogProvider writerProvider = provider(catalog);
+
+ assertThat(readerProvider.listPartitions("year=2025/%")).containsExactly(oldPartition);
+ writerProvider.createPartitions(Collections.singletonList(newPartition.spec()));
+
+ assertThat(readerProvider.listPartitions("year=2025/%"))
+ .containsExactly(oldPartition, newPartition);
+ verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%");
+ }
+
+ @Test
+ void testCreatePartitionsFailureInvalidatesOtherProviderCache() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition oldPartition = partition("2025", "10");
+ Partition newPartition = partition("2025", "11");
+ List storedPartitions = new ArrayList<>();
+ storedPartitions.add(oldPartition);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenAnswer(ignored -> new PagedList<>(new ArrayList<>(storedPartitions), null));
+ List> created = Collections.singletonList(newPartition.spec());
+ RuntimeException responseLost = new RuntimeException("response lost");
+ doAnswer(
+ ignored -> {
+ storedPartitions.add(newPartition);
+ throw responseLost;
+ })
+ .when(catalog)
+ .createPartitions(IDENTIFIER, created);
+ FormatTableCatalogProvider readerProvider = provider(catalog);
+ FormatTableCatalogProvider writerProvider = provider(catalog);
+
+ assertThat(readerProvider.listPartitions("year=2025/%")).containsExactly(oldPartition);
+ assertThatThrownBy(() -> writerProvider.createPartitions(created)).isSameAs(responseLost);
+
+ assertThat(readerProvider.listPartitions("year=2025/%"))
+ .containsExactly(oldPartition, newPartition);
+ verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%");
+ }
+
+ @Test
+ void testAdvanceGenerationInvalidatesProvidersInSameProcess() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition oldPartition = partition("2025", "10");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenReturn(
+ new PagedList<>(Arrays.asList(oldPartition, partition("2025", "11")), null),
+ new PagedList<>(Collections.singletonList(oldPartition), null));
+ FormatTableCatalogProvider provider = provider(catalog);
+
+ assertThat(provider.listPartitions("year=2025/%")).hasSize(2);
+ // The catalog partition DDL path (e.g. RESTCatalog.dropPartitions) advances the
+ // generation so same-process scans read their own writes within the cache TTL.
+ FormatTableCatalogProvider.advanceGeneration(IDENTIFIER);
+
+ assertThat(provider.listPartitions("year=2025/%")).containsExactly(oldPartition);
+ verify(catalog, times(2)).listPartitionsPaged(IDENTIFIER, 1000, null, "year=2025/%");
+ }
+
+ @Test
+ void testLeadingPatternResidualFilterAndUnregisteredDirectory() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), eq("year=2025/%")))
+ .thenReturn(
+ new PagedList<>(
+ Arrays.asList(partition("2025", "10"), partition("2025", "11")),
+ null));
+ TrackingLocalFileIO fileIO = new TrackingLocalFileIO();
+ Path tablePath = new Path(tempDir.toUri());
+ Path octoberFile = writeDataFile(fileIO, tablePath, "year=2025/month=10");
+ Path novemberFile = writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ Path unregisteredFile = writeDataFile(fileIO, tablePath, "year=2025/month=12");
+ FormatTable table = createTable(fileIO, tablePath, provider(catalog), false);
+ PredicateBuilder builder = new PredicateBuilder(table.partitionType());
+ Predicate predicate =
+ PredicateBuilder.and(builder.equal(0, 2025), builder.greaterThan(1, 10));
+ PartitionPredicate filter =
+ PartitionPredicate.fromPredicate(table.partitionType(), predicate);
+
+ ManagedFormatTableScan scan = new ManagedFormatTableScan(table, filter, null);
+ List plannedFiles = plannedFiles(scan.plan().splits());
+
+ assertThat(scan.createPartitionNamePattern()).isEqualTo("year=2025/%");
+ assertThat(plannedFiles).containsExactly(novemberFile);
+ assertThat(plannedFiles).doesNotContain(octoberFile, unregisteredFile);
+ assertThat(fileIO.listedPaths).containsExactly(new Path(tablePath, "year=2025/month=11"));
+ }
+
+ @Test
+ void testListPartitionEntriesUsesCatalogVisibility() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition catalogPartition =
+ new Partition(partition("2025", "11").spec(), 11L, 22L, 3L, 44L, 5, false);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(catalogPartition), null));
+ TrackingLocalFileIO fileIO = new TrackingLocalFileIO();
+ Path tablePath = new Path(tempDir.toUri());
+ writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ writeDataFile(fileIO, tablePath, "year=2025/month=12");
+ FormatTable table = createTable(fileIO, tablePath, provider(catalog), false);
+
+ List entries =
+ new ManagedFormatTableScan(table, null, null).listPartitionEntries();
+
+ assertThat(entries)
+ .extracting(
+ entry -> entry.partition().getInt(0), entry -> entry.partition().getInt(1))
+ .containsExactly(org.assertj.core.groups.Tuple.tuple(2025, 11));
+ assertThat(entries.get(0).recordCount()).isEqualTo(11L);
+ assertThat(entries.get(0).fileSizeInBytes()).isEqualTo(22L);
+ assertThat(entries.get(0).fileCount()).isEqualTo(3L);
+ assertThat(entries.get(0).lastFileCreationTime()).isEqualTo(44L);
+ assertThat(entries.get(0).totalBuckets()).isEqualTo(5);
+ assertThat(fileIO.listedPaths).isEmpty();
+ assertThat(fileIO.statusListedPaths).isEmpty();
+ }
+
+ @Test
+ void testUnderscoreInPartitionNameRemainsLiteralPrefix() {
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table =
+ createStringPartitionTable(fileIO, tablePath, provider(mock(Catalog.class)));
+ Predicate predicate =
+ new PredicateBuilder(table.partitionType())
+ .equal(0, BinaryString.fromString("a_b"));
+ PartitionPredicate filter =
+ PartitionPredicate.fromPredicate(table.partitionType(), predicate);
+
+ ManagedFormatTableScan scan = new ManagedFormatTableScan(table, filter, null);
+
+ assertThat(scan.createPartitionNamePattern()).isEqualTo("year=a_b/%");
+ }
+
+ @Test
+ void testValueOnlyPartitionPath() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(partition("2025", "11")), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = writeDataFile(fileIO, tablePath, "2025/11");
+ FormatTable table = createTable(fileIO, tablePath, provider(catalog), true);
+
+ List plannedFiles =
+ plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ @Test
+ void testDuplicateCatalogPartitionPlansSplitsOnce() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(
+ new PagedList<>(
+ Arrays.asList(partition("2025", "11"), partition("2025", "11")),
+ null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ FormatTable table = createTable(fileIO, tablePath, provider(catalog), false);
+
+ List plannedFiles =
+ plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ @Test
+ void testWhitespacePartitionValueIsVisible() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition whitespacePartition = partition(" ", "11");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(whitespacePartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile =
+ writeDataFile(
+ fileIO,
+ tablePath,
+ PartitionPathUtils.generatePartitionPath(
+ new LinkedHashMap<>(whitespacePartition.spec())));
+ FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog));
+
+ List plannedFiles =
+ plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ @Test
+ void testEmptyPartitionValueReportsCorruptMetadata() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition emptyValuePartition = partition("2025", "");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(emptyValuePartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog));
+
+ assertThatThrownBy(() -> new ManagedFormatTableScan(table, null, null).plan().splits())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("corrupt partition metadata")
+ .hasMessageContaining(IDENTIFIER.getFullName());
+ }
+
+ @Test
+ void testCatalogFailureDoesNotFallBackToFileSystem() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ RuntimeException catalogFailure =
+ new RuntimeException("Catalog partition listing unavailable");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenThrow(catalogFailure);
+ TrackingLocalFileIO fileIO = new TrackingLocalFileIO();
+ Path tablePath = new Path(tempDir.toUri());
+ writeDataFile(fileIO, tablePath, "year=2025/month=11");
+ FormatTable table = createTable(fileIO, tablePath, provider(catalog), false);
+
+ assertThatThrownBy(() -> new ManagedFormatTableScan(table, null, null).plan().splits())
+ .isSameAs(catalogFailure);
+ assertThat(fileIO.listedPaths).isEmpty();
+ }
+
+ @Test
+ @DisplayName(
+ "treats a registered partition with a missing directory as an empty partition "
+ + "(requires real object-store validation)")
+ void testMissingManagedPartitionReadsAsEmpty() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(
+ new PagedList<>(Collections.singletonList(partition("2025", "11")), null));
+ Path tablePath = new Path(tempDir.toUri());
+ Path missingPath = new Path(tablePath, "year=2025/month=11");
+ LocalFileIO fileIO =
+ new LocalFileIO() {
+ @Override
+ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+ assertThat(path).isEqualTo(missingPath);
+ throw new FileNotFoundException(path.toString());
+ }
+ };
+ FormatTable table = createTable(fileIO, tablePath, provider(catalog), false);
+
+ // A registered partition whose directory is missing (e.g. ADD PARTITION before the first
+ // INSERT) must not fail the whole scan; it reads as an empty partition, matching Hive.
+ List plannedFiles =
+ plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits());
+ assertThat(plannedFiles).isEmpty();
+ }
+
+ @Test
+ void testTraversalPartitionValueReportsCorruptMetadata() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition traversalPartition = partition("2025", "..");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(traversalPartition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog), true);
+
+ assertThatThrownBy(() -> new ManagedFormatTableScan(table, null, null).plan().splits())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("corrupt partition metadata")
+ .hasMessageContaining(IDENTIFIER.getFullName());
+ }
+
+ @Test
+ void testDotDotValueInKeyValueLayoutRemainsVisible() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Partition partition = partition("2025", "..");
+ when(catalog.listPartitionsPaged(eq(IDENTIFIER), eq(1000), isNull(), isNull()))
+ .thenReturn(new PagedList<>(Collections.singletonList(partition), null));
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(tempDir.toUri());
+ Path dataFile = writeDataFile(fileIO, tablePath, "year=2025/month=..");
+ FormatTable table = createStringPartitionTable(fileIO, tablePath, provider(catalog));
+
+ List plannedFiles =
+ plannedFiles(new ManagedFormatTableScan(table, null, null).plan().splits());
+
+ assertThat(plannedFiles).containsExactly(dataFile);
+ }
+
+ private FormatTableCatalogProvider provider(Catalog catalog) {
+ return new FormatTableCatalogProvider(IDENTIFIER, () -> catalog);
+ }
+
+ private FormatTable createTable(
+ LocalFileIO fileIO,
+ Path tablePath,
+ FormatTableCatalogProvider provider,
+ boolean valueOnlyPath) {
+ RowType rowType =
+ RowType.builder()
+ .field("year", DataTypes.INT())
+ .field("month", DataTypes.INT())
+ .field("id", DataTypes.INT())
+ .build();
+ return FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(IDENTIFIER)
+ .rowType(rowType)
+ .partitionKeys(Arrays.asList("year", "month"))
+ .location(tablePath.toString())
+ .format(FormatTable.Format.CSV)
+ .options(
+ Collections.singletonMap(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(valueOnlyPath)))
+ .catalogProvider(provider)
+ .build();
+ }
+
+ private FormatTable createStringPartitionTable(
+ LocalFileIO fileIO, Path tablePath, FormatTableCatalogProvider provider) {
+ return createStringPartitionTable(fileIO, tablePath, provider, false);
+ }
+
+ private FormatTable createStringPartitionTable(
+ LocalFileIO fileIO,
+ Path tablePath,
+ FormatTableCatalogProvider provider,
+ boolean valueOnlyPath) {
+ RowType rowType =
+ RowType.builder()
+ .field("year", DataTypes.STRING())
+ .field("month", DataTypes.STRING())
+ .field("id", DataTypes.INT())
+ .build();
+ return FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(IDENTIFIER)
+ .rowType(rowType)
+ .partitionKeys(Arrays.asList("year", "month"))
+ .location(tablePath.toString())
+ .format(FormatTable.Format.CSV)
+ .options(
+ Collections.singletonMap(
+ FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(valueOnlyPath)))
+ .catalogProvider(provider)
+ .build();
+ }
+
+ private Path writeDataFile(LocalFileIO fileIO, Path tablePath, String partitionPath)
+ throws IOException {
+ Path file = new Path(new Path(tablePath, partitionPath), "data.csv");
+ fileIO.mkdirs(file.getParent());
+ try (PositionOutputStream out = fileIO.newOutputStream(file, false)) {
+ out.write(1);
+ }
+ return file;
+ }
+
+ private static Partition partition(String year, String month) {
+ Map spec = new LinkedHashMap<>();
+ spec.put("year", year);
+ spec.put("month", month);
+ return new Partition(spec, 0, 0, 0, 0, -1, false);
+ }
+
+ private static List plannedFiles(List splits) {
+ return splits.stream()
+ .map(FormatDataSplit.class::cast)
+ .flatMap(split -> split.files().stream())
+ .map(FormatDataSplit.FileMeta::filePath)
+ .collect(Collectors.toList());
+ }
+
+ private static class TrackingLocalFileIO extends LocalFileIO {
+
+ private final List listedPaths = new ArrayList<>();
+ private final List statusListedPaths = new ArrayList<>();
+
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ statusListedPaths.add(path);
+ return super.listStatus(path);
+ }
+
+ @Override
+ public FileStatus[] listFiles(Path path, boolean recursive) throws IOException {
+ listedPaths.add(path);
+ return super.listFiles(path, recursive);
+ }
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
index a21b168f54ea..6b2952778003 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/PartitionPathUtilsTest.java
@@ -19,17 +19,25 @@
package org.apache.paimon.utils;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import java.util.Arrays;
+import java.util.LinkedHashMap;
import static org.apache.paimon.utils.PartitionPathUtils.mightMatch;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.entry;
-/** Tests for {@link PartitionPathUtils#mightMatch}. */
+/** Tests for {@link PartitionPathUtils}. */
class PartitionPathUtilsTest {
private final RowType partitionType =
@@ -39,6 +47,46 @@ class PartitionPathUtilsTest {
.build();
private final PredicateBuilder builder = new PredicateBuilder(partitionType);
+ @ParameterizedTest
+ @ValueSource(strings = {".", ".."})
+ void testValueOnlyDotSegmentIsRejected(String rawValue) {
+ LinkedHashMap partitionSpec = new LinkedHashMap<>();
+ partitionSpec.put("pt", rawValue);
+
+ assertThatThrownBy(() -> PartitionPathUtils.generatePartitionPathUtil(partitionSpec, true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(rawValue);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {".", ".."})
+ void testKeyedDotValueKeepsCompatibleSafeLayout(String rawValue) {
+ LinkedHashMap partitionSpec = new LinkedHashMap<>();
+ partitionSpec.put("pt", rawValue);
+
+ String partitionPath = PartitionPathUtils.generatePartitionPathUtil(partitionSpec, false);
+ Path resolvedPath = new Path(new Path("file:///warehouse/table"), partitionPath);
+
+ assertThat(partitionPath).isEqualTo("pt=" + rawValue + Path.SEPARATOR);
+ assertThat(PartitionPathUtils.extractPartitionSpecFromPath(resolvedPath))
+ .containsExactly(entry("pt", rawValue));
+ }
+
+ @Test
+ void testTrailingPartitionExtractionStopsAtTableBoundary() {
+ Path partitionPath = new Path("file:///warehouse/env=prod/table/dt=20260718/hh=10");
+
+ assertThat(
+ PartitionPathUtils.extractPartitionSpecFromPath(
+ partitionPath, Arrays.asList("dt", "hh")))
+ .containsExactly(entry("dt", "20260718"), entry("hh", "10"));
+ assertThat(
+ PartitionPathUtils.extractPartitionSpecFromPath(
+ new Path("file:///warehouse/dt=parent/table/wrong=20260718/hh=10"),
+ Arrays.asList("dt", "hh")))
+ .isNull();
+ }
+
@Test
void testNullPredicate() {
assertThat(mightMatch(null, 0, 0, GenericRow.of(2024, 5))).isTrue();
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
index 853c7f41467b..bec591bbd7d1 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/FormatTableBatchWriteBase.scala
@@ -49,6 +49,8 @@ abstract class FormatTableBatchWriteBase(
extends Logging
with Serializable {
+ @volatile protected var commitStarted: Boolean = false
+
protected val batchWriteBuilder: BatchWriteBuilder = {
val builder = table.newBatchWriteBuilder()
// todo: add test for static overwrite the whole table
@@ -65,6 +67,7 @@ abstract class FormatTableBatchWriteBase(
}
protected def commitMessages(messages: Array[WriterCommitMessage]): Unit = {
+ commitStarted = true
logInfo(s"Committing to FormatTable ${table.name()}")
val batchTableCommit = batchWriteBuilder.newCommit()
val commitMessages = WriteTaskResult.merge(messages).asJava
@@ -80,6 +83,12 @@ abstract class FormatTableBatchWriteBase(
}
protected def abortMessages(messages: Array[WriterCommitMessage]): Unit = {
+ if (commitStarted) {
+ logWarning(
+ s"Skip abort cleanup for FormatTable ${table.name()} because commit has already started")
+ return
+ }
+
logInfo(s"Aborting write to FormatTable ${table.name()}")
val batchTableCommit = batchWriteBuilder.newCommit()
val commitMessages = WriteTaskResult.merge(messages).asJava
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala
new file mode 100644
index 000000000000..ce4d1951e1b4
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTableBatchWriteTest.scala
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.format
+
+import org.apache.paimon.spark.write.FormatTableWriteTaskResult
+import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.sink.{BatchTableCommit, BatchWriteBuilder}
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.connector.write.WriterCommitMessage
+import org.apache.spark.sql.types.StructType
+
+import java.lang.reflect.{InvocationHandler, Method, Proxy}
+import java.util.concurrent.atomic.AtomicInteger
+
+class FormatTableBatchWriteTest extends SparkFunSuite {
+
+ test("abort after commit starts does not clean up potentially committed files") {
+ val commitCalls = new AtomicInteger
+ val abortCalls = new AtomicInteger
+ val batchWrite =
+ new FormatTableBatchWrite(formatTable(commitCalls, abortCalls), None, None, StructType(Nil))
+ val messages: Array[WriterCommitMessage] = Array(FormatTableWriteTaskResult(Seq.empty))
+
+ val error = intercept[RuntimeException] {
+ batchWrite.commit(messages)
+ }
+ assert(error.getMessage == "partition registration response lost")
+
+ // Spark invokes abort after commit throws. The commit outcome may be ambiguous, so cleanup
+ // could delete files which were already committed and registered remotely.
+ batchWrite.abort(messages)
+
+ assert(commitCalls.get == 1)
+ assert(abortCalls.get == 0)
+ }
+
+ test("abort before commit delegates cleanup") {
+ val commitCalls = new AtomicInteger
+ val abortCalls = new AtomicInteger
+ val batchWrite =
+ new FormatTableBatchWrite(formatTable(commitCalls, abortCalls), None, None, StructType(Nil))
+
+ batchWrite.abort(Array(FormatTableWriteTaskResult(Seq.empty)))
+
+ assert(commitCalls.get == 0)
+ assert(abortCalls.get == 1)
+ }
+
+ private def formatTable(commitCalls: AtomicInteger, abortCalls: AtomicInteger): FormatTable = {
+ val tableCommit = proxy(classOf[BatchTableCommit]) {
+ case "commit" =>
+ commitCalls.incrementAndGet()
+ throw new RuntimeException("partition registration response lost")
+ case "abort" =>
+ abortCalls.incrementAndGet()
+ null
+ }
+ val writeBuilder = proxy(classOf[BatchWriteBuilder]) { case "newCommit" => tableCommit }
+ proxy(classOf[FormatTable]) {
+ case "newBatchWriteBuilder" => writeBuilder
+ case "name" => "test_db.format_table"
+ }
+ }
+
+ private def proxy[T](clazz: Class[T])(responses: PartialFunction[String, AnyRef]): T = {
+ Proxy
+ .newProxyInstance(
+ clazz.getClassLoader,
+ Array(clazz),
+ new InvocationHandler {
+ override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = {
+ responses.applyOrElse(
+ method.getName,
+ (name: String) =>
+ throw new UnsupportedOperationException(s"Unexpected $name invocation"))
+ }
+ }
+ )
+ .asInstanceOf[T]
+ }
+}