You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/07/21 07:50:35 UTC
(paimon) branch master updated: [spark] Support partition DDL for catalog-managed Format Table partitions (#8751)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new d4d1c01157 [spark] Support partition DDL for catalog-managed Format Table partitions (#8751)
d4d1c01157 is described below
commit d4d1c01157937d4622892f661a41bd358d7be658
Author: Dapeng Sun(孙大鹏) <da...@alibaba-inc.com>
AuthorDate: Tue Jul 21 15:50:29 2026 +0800
[spark] Support partition DDL for catalog-managed Format Table partitions (#8751)
---
docs/docs/spark/sql-ddl.md | 49 ++
.../spark/format/FormatTablePartitionRepair.java | 165 ++++
.../paimon/spark/PaimonPartitionManagement.scala | 56 +-
.../PaimonFormatTablePartitionDdlExec.scala | 194 +++++
.../paimon/spark/execution/PaimonStrategy.scala | 48 +-
.../paimon/spark/format/PaimonFormatTable.scala | 193 ++++-
.../extensions/RewriteSparkDDLCommands.scala | 36 +-
.../extensions/UnresolvedPaimonRelation.scala | 49 --
.../format/FormatTablePartitionRepairTest.java | 468 ++++++++++
.../apache/paimon/spark/util/OptionUtilsTest.scala | 135 +++
.../FormatTablePartitionDdlPlanningTest.scala | 951 +++++++++++++++++++++
.../format/CatalogManagedPartitionLoadTest.scala | 46 +
.../FormatTablePartitionManagementTest.scala | 768 +++++++++++++++++
.../CatalogManagedPartitionMsckRepairTest.scala | 669 +++++++++++++++
14 files changed, 3737 insertions(+), 90 deletions(-)
diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md
index aa9310b8b4..4ac28fd652 100644
--- a/docs/docs/spark/sql-ddl.md
+++ b/docs/docs/spark/sql-ddl.md
@@ -206,6 +206,55 @@ CREATE TABLE my_table (
);
```
+### Manage Format Table Partitions
+
+For a Format Table whose `metastore.partitioned-table` option is `true`, the catalog holds the
+partitions and Spark supports the standard partition DDL:
+
+```sql
+ALTER TABLE my_table ADD PARTITION (dt='2025-01-01');
+ALTER TABLE my_table DROP PARTITION (dt='2025-01-01');
+MSCK REPAIR TABLE my_table;
+SHOW PARTITIONS my_table;
+```
+
+On a Format Table whose partitions are discovered from the filesystem, `ADD PARTITION`,
+`DROP PARTITION` and `MSCK REPAIR TABLE` fail with an error.
+
+`ADD PARTITION` creates the partition directory and registers the partition; querying a newly
+added partition before any data is written returns no rows. `DROP PARTITION` unregisters the
+partition and deletes its directory.
+
+:::info
+
+`metastore.partitioned-table = true` enables catalog-managed partitions, which requires an
+internal Format Table in a catalog that supports it (currently the REST catalog) and cannot be
+combined with `format-table.implementation = engine`. The REST catalog validates this
+combination on `CREATE TABLE`, with catalog-level table defaults
+(`spark.sql.catalog.paimon.table-default.*`) participating in the effective options: a default
+that makes the combination invalid fails the DDL.
+
+In a REST catalog, asking for catalog-managed partitions on a table that cannot have them — an
+external table, or `format-table.implementation = engine` — fails, rather than handing back a
+table whose options say one thing and whose partitions come from somewhere else. Remove the option
+with `ALTER TABLE my_table UNSET TBLPROPERTIES ('metastore.partitioned-table')`.
+
+In any other catalog the option keeps the meaning it has always had on a Format Table — none. Such
+a table loads and reads its partitions from the filesystem, so on a Format Table it
+only takes effect in a REST catalog; elsewhere partitions are discovered from the filesystem.
+On a REST catalog, an existing Format Table whose partitions were never registered reads as empty
+until you register them with `MSCK REPAIR TABLE my_table`.
+
+Mixed-version note: only writers that support catalog-managed partitions register the partitions
+they produce. During a rolling upgrade, upgrade all writers before relying on catalog-managed
+partitions, or run `MSCK REPAIR TABLE` afterwards, since data written by an older writer is not
+visible until its partitions are registered.
+
+For a Format Table, `metastore.partitioned-table` only changes where partitions come from; it does not
+change the table's managed/external ownership.
+
+:::
+
### Create External Table
When the catalog's `metastore` type is `hive`, if the `location` is specified when creating a table, that table will be considered an external table; otherwise, it will be a managed table.
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java
new file mode 100644
index 0000000000..4aa99b812c
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/format/FormatTablePartitionRepair.java
@@ -0,0 +1,165 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
+import org.apache.paimon.utils.Pair;
+import org.apache.paimon.utils.PartitionPathUtils;
+import org.apache.paimon.utils.Preconditions;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Repairs a Format Table with catalog-managed partitions by reconciling its partition directories
+ * with its catalog registration, backing {@code MSCK REPAIR TABLE [{ADD|DROP|SYNC} PARTITIONS]}.
+ * Only partition metadata is changed, data files are never touched.
+ */
+public class FormatTablePartitionRepair {
+
+ private FormatTablePartitionRepair() {}
+
+ /**
+ * Repair the partition metadata of a Format Table with catalog-managed partitions. The flag
+ * pair follows Spark's {@code RepairTable} plan (plain MSCK means ADD). Returns the number of
+ * applied actions.
+ */
+ public static int repair(
+ PaimonFormatTable sparkTable, boolean addPartitions, boolean dropPartitions) {
+ Preconditions.checkArgument(
+ addPartitions || dropPartitions,
+ "MSCK REPAIR TABLE must enable ADD and/or DROP partitions");
+ FormatTable formatTable = sparkTable.table();
+ FormatTablePartitionManager partitionManager = formatTable.partitionManager();
+ Preconditions.checkArgument(
+ partitionManager != null,
+ "%s does not have catalog-managed partitions",
+ formatTable.fullName());
+
+ return apply(
+ partitionManager,
+ listFilesystemPartitionSpecs(formatTable),
+ formatTable.partitionKeys(),
+ addPartitions,
+ dropPartitions);
+ }
+
+ private static List<Map<String, String>> listFilesystemPartitionSpecs(FormatTable formatTable) {
+ // Discover partitions from the raw directory names rather than through the table scan:
+ // the scan casts each value to its column type and back (e.g. month=01 -> 1), producing
+ // specs that can no longer round-trip to the real directory. The write path registers the
+ // raw directory value, so a repair must diff against the same raw values to avoid
+ // spuriously adding/dropping partition metadata.
+ boolean onlyValueInPath =
+ new CoreOptions(formatTable.options()).formatTablePartitionOnlyValueInPath();
+ List<Pair<LinkedHashMap<String, String>, Path>> found =
+ PartitionPathUtils.searchPartSpecAndPaths(
+ formatTable.fileIO(),
+ new Path(formatTable.location()),
+ formatTable.partitionKeys().size(),
+ formatTable.partitionKeys(),
+ onlyValueInPath);
+ List<Map<String, String>> specs = new ArrayList<>(found.size());
+ for (Pair<LinkedHashMap<String, String>, Path> pair : found) {
+ PartitionPathUtils.validatePartitionSpecForPath(pair.getKey(), onlyValueInPath);
+ specs.add(pair.getKey());
+ }
+ return specs;
+ }
+
+ /**
+ * Diff the filesystem partition set against the catalog registration set and apply the
+ * requested actions. ADD registers "directory exists but unregistered"; DROP is metadata-only
+ * cleanup of "registered but directory missing". Scan-completeness guard: the filesystem
+ * listing that feeds {@code filesystemPartitions} ({@link
+ * PartitionPathUtils#searchPartSpecAndPaths}) fails on any mid-scan LIST error instead of
+ * returning a truncated set, so a DROP diff can only be produced from a complete listing and a
+ * transient failure never deregisters partitions that still exist.
+ */
+ static int apply(
+ FormatTablePartitionManager partitionManager,
+ List<Map<String, String>> filesystemPartitions,
+ List<String> partitionKeys,
+ boolean addPartitions,
+ boolean dropPartitions) {
+ Set<Map<String, String>> registeredPartitions = new HashSet<>();
+ for (Partition partition : partitionManager.listPartitions(Collections.emptyMap())) {
+ registeredPartitions.add(partition.spec());
+ }
+
+ Set<Map<String, String>> filesystemSet = new HashSet<>(filesystemPartitions);
+
+ List<Map<String, String>> addDiff = new ArrayList<>();
+ if (addPartitions) {
+ for (Map<String, String> partition : filesystemPartitions) {
+ if (!registeredPartitions.contains(partition)) {
+ addDiff.add(partition);
+ }
+ }
+ sortByCanonicalPath(addDiff, partitionKeys);
+ }
+ List<Map<String, String>> dropDiff = new ArrayList<>();
+ if (dropPartitions) {
+ for (Map<String, String> partition : registeredPartitions) {
+ if (!filesystemSet.contains(partition)) {
+ dropDiff.add(partition);
+ }
+ }
+ sortByCanonicalPath(dropDiff, partitionKeys);
+ }
+
+ // A first repair of a pre-existing table can discover far more partitions than any regular
+ // write. Splitting such a diff into per-request batches is the partition catalog's job;
+ // registration is an idempotent upsert and unregistration ignores missing partitions, so a
+ // mid-way failure leaves a state a rerun converges from.
+ if (!addDiff.isEmpty()) {
+ partitionManager.createPartitions(addDiff, true);
+ }
+ if (!dropDiff.isEmpty()) {
+ partitionManager.dropPartitions(dropDiff);
+ }
+ return addDiff.size() + dropDiff.size();
+ }
+
+ /** Sort partitions by their canonical path for a stable, deterministic apply order. */
+ private static void sortByCanonicalPath(
+ List<Map<String, String>> partitions, List<String> partitionKeys) {
+ partitions.sort(Comparator.comparing(partition -> canonicalPath(partition, partitionKeys)));
+ }
+
+ private static String canonicalPath(Map<String, String> partition, List<String> partitionKeys) {
+ LinkedHashMap<String, String> orderedSpec = new LinkedHashMap<>();
+ for (String key : partitionKeys) {
+ if (partition.containsKey(key)) {
+ orderedSpec.put(key, partition.get(key));
+ }
+ }
+ return PartitionPathUtils.generatePartitionPath(orderedSpec);
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
index cd5955b0ff..3f847c1252 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/PaimonPartitionManagement.scala
@@ -46,49 +46,43 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement with L
private def toPaimonPartitions(rows: Array[InternalRow]): Array[java.util.Map[String, String]] = {
table match {
- case fileStoreTable: FileStoreTable =>
+ case _: FileStoreTable =>
val partitionKeys = table.partitionKeys().asScala.toSeq
- val partitionDefaultName = fileStoreTable.coreOptions().partitionDefaultName()
- val legacyPartitionName = CoreOptions.fromMap(table.options()).legacyPartitionName
-
rows.map {
r =>
- val partitionFieldCount = r.numFields
require(
- partitionFieldCount <= partitionKeys.length,
- s"Partition values length $partitionFieldCount exceeds partition keys " +
+ r.numFields <= partitionKeys.length,
+ s"Partition values length ${r.numFields} exceeds partition keys " +
s"${partitionKeys.mkString("[", ", ", "]")}."
)
- val partitionNames = partitionKeys.take(partitionFieldCount)
- val currentPartitionRowType =
- if (partitionFieldCount == partitionRowType.getFieldCount) {
- partitionRowType
- } else {
- TypeUtils.project(table.rowType, partitionNames.asJava)
- }
- val currentPartitionSchema =
- if (partitionFieldCount == partitionSchema.length) {
- partitionSchema
- } else {
- SparkTypeUtils.fromPaimonRowType(currentPartitionRowType)
- }
- val rowConverter = CatalystTypeConverters.createToScalaConverter(
- CharVarcharUtils.replaceCharVarcharWithString(currentPartitionSchema))
- val rowDataPartitionComputer = new InternalRowPartitionComputer(
- partitionDefaultName,
- currentPartitionRowType,
- partitionNames.toArray,
- legacyPartitionName
- )
-
- rowDataPartitionComputer.generatePartValues(
- new SparkRow(currentPartitionRowType, rowConverter(r).asInstanceOf[Row]))
+ toPaimonPartition(r, partitionKeys.take(r.numFields))
}
case _ =>
throw new UnsupportedOperationException("Only FileStoreTable supports partitions.")
}
}
+ protected def toPaimonPartition(
+ row: InternalRow,
+ partitionNames: Seq[String]): java.util.Map[String, String] = {
+ val coreOptions = CoreOptions.fromMap(table.options())
+ val partitionDefaultName = coreOptions.partitionDefaultName()
+ val legacyPartitionName = coreOptions.legacyPartitionName
+ val currentPartitionRowType = TypeUtils.project(table.rowType, partitionNames.asJava)
+ val currentPartitionSchema = SparkTypeUtils.fromPaimonRowType(currentPartitionRowType)
+ val rowConverter = CatalystTypeConverters.createToScalaConverter(
+ CharVarcharUtils.replaceCharVarcharWithString(currentPartitionSchema))
+ val rowDataPartitionComputer = new InternalRowPartitionComputer(
+ partitionDefaultName,
+ currentPartitionRowType,
+ partitionNames.toArray,
+ legacyPartitionName
+ )
+
+ rowDataPartitionComputer.generatePartValues(
+ new SparkRow(currentPartitionRowType, rowConverter(row).asInstanceOf[Row]))
+ }
+
override def dropPartitions(rows: Array[InternalRow]): Boolean = {
table match {
case fileStoreTable: FileStoreTable =>
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala
new file mode 100644
index 0000000000..a49dab7019
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonFormatTablePartitionDdlExec.scala
@@ -0,0 +1,194 @@
+/*
+ * 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.execution
+
+import org.apache.paimon.spark.format.{FormatTablePartitionRepair, PaimonFormatTable}
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionsException, ResolvedPartitionSpec}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.execution.datasources.v2.LeafV2CommandExec
+
+import java.util.{Map => JMap}
+
+import scala.collection.JavaConverters._
+
+object PaimonFormatTablePartitionDdlExec {
+
+ /**
+ * Run an operation and refresh Spark's cached plans afterwards, whether it succeeded or not. A
+ * refresh failure never replaces the operation's own failure — that one explains what went wrong
+ * — and is attached to it instead, unless the two are the same throwable.
+ */
+ private[execution] def refreshingCache[T](refreshCache: () => Unit)(operation: => T): T = {
+ var operationFailure: Throwable = null
+ try {
+ operation
+ } catch {
+ case failure: Throwable =>
+ operationFailure = failure
+ throw failure
+ } finally {
+ try {
+ refreshCache()
+ } catch {
+ case refreshFailure: Throwable =>
+ if (operationFailure == null) {
+ throw refreshFailure
+ } else if (refreshFailure ne operationFailure) {
+ operationFailure.addSuppressed(refreshFailure)
+ }
+ }
+ }
+ }
+
+ private[execution] def unsupportedWithoutCatalogManagedPartitions(
+ operation: String,
+ table: PaimonFormatTable): UnsupportedOperationException =
+ new UnsupportedOperationException(
+ s"$operation PARTITION is supported only for a Format Table with catalog-managed " +
+ s"partitions, but partitions of ${table.name()} are discovered from the filesystem.")
+}
+
+/**
+ * Registers catalog-managed partitions of a Format Table without Spark's client-side existence
+ * precheck. The whole batch and IF NOT EXISTS flag are forwarded to the catalog so it can apply
+ * them atomically.
+ */
+case class PaimonAddFormatTablePartitionsExec(
+ table: PaimonFormatTable,
+ partSpecs: Seq[ResolvedPartitionSpec],
+ ignoreIfExists: Boolean,
+ refreshCache: () => Unit)
+ extends LeafV2CommandExec {
+
+ override protected def run(): Seq[InternalRow] = {
+ if (!table.hasCatalogManagedPartitions) {
+ throw PaimonFormatTablePartitionDdlExec.unsupportedWithoutCatalogManagedPartitions(
+ "ADD",
+ table)
+ }
+
+ if (partSpecs.nonEmpty) {
+ val properties: Array[JMap[String, String]] =
+ partSpecs.map(spec => spec.location.map("location" -> _).toMap.asJava).toArray
+ PaimonFormatTablePartitionDdlExec.refreshingCache(refreshCache) {
+ table.createFormatTablePartitions(
+ partSpecs.map(_.ident).toArray,
+ properties,
+ ignoreIfExists)
+ }
+ }
+ Seq.empty
+ }
+
+ override def output: Seq[Attribute] = Seq.empty
+}
+
+/**
+ * Physical command providing explicit DROP semantics for Format Tables with catalog-managed
+ * partitions (tables using filesystem partition discovery always fail with a clear error). Complete
+ * specs are resolved against the catalog registration first: missing specs fail with
+ * [[NoSuchPartitionsException]] unless IF EXISTS was given, and only registered specs are
+ * unregistered and have their directories deleted, so data that is merely awaiting registration
+ * (e.g. by MSCK REPAIR TABLE) is never removed. Partial specs resolve to the registered leaf
+ * partitions they cover.
+ */
+case class PaimonDropFormatTablePartitionsExec(
+ table: PaimonFormatTable,
+ partSpecs: Seq[ResolvedPartitionSpec],
+ ifExists: Boolean,
+ purge: Boolean,
+ refreshCache: () => Unit)
+ extends LeafV2CommandExec {
+
+ override protected def run(): Seq[InternalRow] = {
+ if (!table.hasCatalogManagedPartitions) {
+ throw PaimonFormatTablePartitionDdlExec.unsupportedWithoutCatalogManagedPartitions(
+ "DROP",
+ table)
+ }
+ if (purge) {
+ // Match Spark's v2 default for purgePartitions. DROP PARTITION on a Format Table with
+ // catalog-managed partitions already removes the partition data, so PURGE adds nothing.
+ throw new UnsupportedOperationException(
+ s"DROP PARTITION ... PURGE is not supported for Format Table ${table.name()}. " +
+ "DROP PARTITION already removes the partition data.")
+ }
+ if (partSpecs.isEmpty) {
+ return Seq.empty
+ }
+
+ val partitionKeyCount = table.table.partitionKeys().size()
+ val (completeSpecs, partialSpecs) =
+ partSpecs.partition(_.ident.numFields == partitionKeyCount)
+ val registration = table
+ .formatTablePartitionsRegistered(
+ completeSpecs.map(_.names.toArray).toArray,
+ completeSpecs.map(_.ident).toArray)
+ .toSeq
+ val missingSpecs = completeSpecs.zip(registration).collect { case (spec, false) => spec }
+ if (missingSpecs.nonEmpty && !ifExists) {
+ throw new NoSuchPartitionsException(
+ table.name(),
+ missingSpecs.map(_.ident),
+ table.partitionSchema)
+ }
+
+ val specsToDrop =
+ completeSpecs.zip(registration).collect { case (spec, true) => spec } ++ partialSpecs
+ if (specsToDrop.nonEmpty) {
+ // Catalog-managed semantics (PaimonPartitionManagement#dropFormatTablePartitions): resolve
+ // partial specs, unregister the exact catalog partitions, then delete their directories
+ // with the table FileIO client-side. A directory-deletion failure stays unregistered so
+ // partially deleted data is not exposed again.
+ PaimonFormatTablePartitionDdlExec.refreshingCache(refreshCache) {
+ table.dropFormatTablePartitions(
+ specsToDrop.map(_.names.toArray).toArray,
+ specsToDrop.map(_.ident).toArray)
+ }
+ }
+ Seq.empty
+ }
+
+ override def output: Seq[Attribute] = Seq.empty
+}
+
+/**
+ * MSCK REPAIR TABLE for Format Tables with catalog-managed partitions. Spark rejects RepairTable
+ * for v2 tables in DataSourceV2Strategy, so PaimonStrategy intercepts first; plain MSCK means ADD.
+ * Tables using filesystem partition discovery are not intercepted and keep Spark's own v2
+ * rejection.
+ */
+case class PaimonRepairFormatTablePartitionsExec(
+ table: PaimonFormatTable,
+ addPartitions: Boolean,
+ dropPartitions: Boolean,
+ refreshCache: () => Unit)
+ extends LeafV2CommandExec {
+
+ override protected def run(): Seq[InternalRow] = {
+ PaimonFormatTablePartitionDdlExec.refreshingCache(refreshCache) {
+ FormatTablePartitionRepair.repair(table, addPartitions, dropPartitions)
+ }
+ Seq.empty
+ }
+
+ override def output: Seq[Attribute] = Seq.empty
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
index 74c0e6c47d..2686ef7d76 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/execution/PaimonStrategy.scala
@@ -29,6 +29,7 @@ import org.apache.paimon.spark.catalog.{SparkBaseCatalog, SupportView}
import org.apache.paimon.spark.catalyst.analysis.ResolvedPaimonView
import org.apache.paimon.spark.catalyst.plans.logical.{CopyIntoLocationCommand, CopyIntoLocationSource, CopyIntoTableCommand, CreateOrReplaceTagCommand, CreatePaimonView, DeleteTagCommand, DropPaimonView, LateralVectorSearch, PaimonCallCommand, PaimonDropPartitions, PaimonTableValuedFunctions, RenameTagCommand, ResolvedIdentifier, ShowPaimonViews, ShowTagsCommand, TruncatePaimonTableWithFilter}
import org.apache.paimon.spark.data.SparkInternalRow
+import org.apache.paimon.spark.format.PaimonFormatTable
import org.apache.paimon.spark.read.VectorSearchResultUtils
import org.apache.paimon.spark.schema.PaimonMetadataColumn
import org.apache.paimon.table.{InnerTable, SpecialFields, Table}
@@ -42,7 +43,7 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{ResolvedNamespace, ResolvedTable}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, Expression, GenericInternalRow, JoinedRow, PredicateHelper, UnsafeProjection}
-import org.apache.spark.sql.catalyst.plans.logical.{CreateTableAsSelect, DescribeRelation, LogicalPlan, ReplaceTable, ReplaceTableAsSelect, ShowCreateTable}
+import org.apache.spark.sql.catalyst.plans.logical.{AddPartitions, CreateTableAsSelect, DescribeRelation, DropPartitions, LogicalPlan, RepairTable, ReplaceTable, ReplaceTableAsSelect, ShowCreateTable}
import org.apache.spark.sql.catalyst.util.ArrayData
import org.apache.spark.sql.connector.catalog.{Identifier, PaimonLookupCatalog, TableCatalog}
import org.apache.spark.sql.execution.{PaimonDescribeTableExec, SparkPlan, SparkStrategy}
@@ -161,6 +162,51 @@ case class PaimonStrategy(spark: SparkSession)
case _ => Nil
}
+ case AddPartitions(r @ ResolvedTable(_, _, table: PaimonFormatTable, _), parts, ifNotExists) =>
+ PaimonAddFormatTablePartitionsExec(
+ table,
+ parts.asResolvedPartitionSpecs,
+ ifNotExists,
+ recacheTable(r)) :: Nil
+
+ // Spark's DataSourceV2Strategy rejects RepairTable for every v2 table; Format Tables with
+ // catalog-managed partitions support it through the sync engine, so intercept here (extension
+ // strategies run first). Tables using filesystem partition discovery fall through and keep
+ // the upstream rejection.
+ case RepairTable(
+ r @ ResolvedTable(_, _, table: PaimonFormatTable, _),
+ enableAddPartitions,
+ enableDropPartitions) if table.hasCatalogManagedPartitions =>
+ PaimonRepairFormatTablePartitionsExec(
+ table,
+ enableAddPartitions,
+ enableDropPartitions,
+ recacheTable(r)) :: Nil
+
+ case DropPartitions(
+ r @ ResolvedTable(_, _, table: PaimonFormatTable, _),
+ parts,
+ ifExists,
+ purge) =>
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ parts.asResolvedPartitionSpecs,
+ ifExists,
+ purge,
+ recacheTable(r)) :: Nil
+
+ case PaimonDropPartitions(
+ r @ ResolvedTable(_, _, table: PaimonFormatTable, _),
+ parts,
+ ifExists,
+ purge) =>
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ parts.asResolvedPartitionSpecs,
+ ifExists,
+ purge,
+ recacheTable(r)) :: Nil
+
case PaimonDropPartitions(
r @ ResolvedTable(_, _, table: SparkTable, _),
parts,
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
index dc7effd39f..776fc98ca1 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/format/PaimonFormatTable.scala
@@ -18,12 +18,17 @@
package org.apache.paimon.spark.format
+import org.apache.paimon.CoreOptions
import org.apache.paimon.format.csv.CsvOptions
+import org.apache.paimon.fs.Path
import org.apache.paimon.spark.{BaseTable, FormatTableScanBuilder}
import org.apache.paimon.spark.write.BaseV2WriteBuilder
import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.format.FormatTablePartitionManager
import org.apache.paimon.types.RowType
+import org.apache.paimon.utils.PartitionPathUtils
+import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite, TableCapability, TableCatalog}
import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ, BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC}
import org.apache.spark.sql.connector.read.ScanBuilder
@@ -34,13 +39,27 @@ import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import java.util
-import java.util.Locale
+import java.util.{Collections, Locale, Map => JMap, Objects}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.{ArrayBuffer, HashSet}
case class PaimonFormatTable(table: FormatTable)
extends BaseTable
with SupportsRead
with SupportsWrite {
+ // A Format Table uses catalog-managed partitions exactly when the catalog gave it a partition
+ // manager; tables using filesystem partition discovery return null and rely on the directory
+ // layout instead.
+ private[spark] def partitionManager: FormatTablePartitionManager = table.partitionManager()
+
+ /**
+ * A Format Table uses catalog-managed partitions exactly when the catalog gave it a partition
+ * manager; otherwise its partitions are discovered from the filesystem directory layout.
+ */
+ def hasCatalogManagedPartitions: Boolean = partitionManager != null
+
override def capabilities(): util.Set[TableCapability] = {
util.EnumSet.of(BATCH_READ, BATCH_WRITE, OVERWRITE_DYNAMIC, OVERWRITE_BY_FILTER)
}
@@ -70,6 +89,178 @@ case class PaimonFormatTable(table: FormatTable)
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
PaimonFormatTableWriterBuilder(table, info.schema)
}
+
+ /**
+ * Resolves, with a single catalog list-by-names lookup, which of the given complete partition
+ * specs are registered. The result is aligned with the input arrays.
+ */
+ private[spark] def formatTablePartitionsRegistered(
+ partitionNames: Array[Array[String]],
+ rows: Array[InternalRow]): Array[Boolean] = {
+ if (rows.isEmpty) {
+ return Array.empty
+ }
+ val requested =
+ rows.zip(partitionNames).map { case (row, names) => toPaimonPartition(row, names.toSeq) }
+ val registered = requirePartitionManager().listPartitionsByNames(requested.toSeq.asJava)
+ val registeredSpecs = registered.asScala.map(_.spec().asScala.toMap).toSet
+ requested.map(spec => registeredSpecs.contains(spec.asScala.toMap))
+ }
+
+ private[spark] def createFormatTablePartitions(
+ rows: Array[InternalRow],
+ maps: Array[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ if (maps.exists(_.keySet().asScala.exists(_.equalsIgnoreCase("location")))) {
+ throw new UnsupportedOperationException(
+ s"ADD PARTITION with LOCATION is not supported for Format Table ${table.fullName()}.")
+ }
+ val onlyValueInPath =
+ CoreOptions.fromMap(table.options()).formatTablePartitionOnlyValueInPath()
+ val partitionKeys = table.partitionKeys().asScala.toSeq
+ val specs = rows.map(row => toPaimonPartition(row, partitionKeys.take(row.numFields))).toSeq
+ // Resolve (and path-safety validate) every directory before mutating anything.
+ val partitionPaths =
+ specs.map(spec => resolvePartitionPathWithinTable(orderedSpec(spec), onlyValueInPath))
+ requirePartitionManager().createPartitions(specs.asJava, ignoreIfExists)
+ // Create the partition directories client-side (symmetric with DROP deleting them), so an
+ // added partition exists on the filesystem and a subsequent scan returns an empty partition
+ // rather than depending on lazy directory creation, matching Hive ADD PARTITION semantics.
+ val fileIO = table.fileIO()
+ partitionPaths.foreach(partitionPath => fileIO.mkdirs(partitionPath))
+ }
+
+ /**
+ * Drops the given partitions: complete specs are unregistered and their directories deleted
+ * as-is, partial specs are expanded to the registered leaf partitions they cover. Callers are
+ * responsible for resolving which complete specs are actually registered first (see
+ * [[formatTablePartitionsRegistered]]), so unregistered data directories are never deleted.
+ */
+ private[spark] def dropFormatTablePartitions(
+ partitionNames: Array[Array[String]],
+ rows: Array[InternalRow]): Boolean = {
+ val partitionKeyCount = table.partitionKeys().size()
+ val requested =
+ rows.zip(partitionNames).map { case (row, names) => toPaimonPartition(row, names.toSeq) }
+ val partitions = ArrayBuffer.empty[JMap[String, String]]
+ val seenPartitions = HashSet.empty[Map[String, String]]
+
+ def addPartition(partition: JMap[String, String]): Unit = {
+ if (seenPartitions.add(partition.asScala.toMap)) {
+ partitions += partition
+ }
+ }
+
+ // Preserve exact requests as-is and let discovery add only missing complete leaves.
+ requested.filter(_.size() == partitionKeyCount).foreach(addPartition)
+ val partialSpecs = requested.filter(_.size() < partitionKeyCount).toSeq.distinct
+ if (partialSpecs.nonEmpty) {
+ def matchesRequestedPartial(partition: JMap[String, String]): Boolean = {
+ partialSpecs.exists(_.asScala.forall {
+ case (key, value) => Objects.equals(value, partition.get(key))
+ })
+ }
+
+ // One unfiltered traversal resolves every partial spec; the requested constraints are
+ // enforced client-side.
+ requirePartitionManager()
+ .listPartitions(Collections.emptyMap[String, String]())
+ .asScala
+ .foreach {
+ partition =>
+ val validated = validateCatalogRegisteredPartition(partition.spec())
+ if (matchesRequestedPartial(validated)) {
+ addPartition(validated)
+ }
+ }
+ }
+ dropCatalogRegisteredPartitions(partitions.toSeq)
+ }
+
+ private def dropCatalogRegisteredPartitions(partitions: Seq[JMap[String, String]]): Boolean = {
+ // Unregister first so new queries stop seeing the partition, then delete the data directory
+ // with the table FileIO (client-side; the server never deletes data). A deletion failure leaves
+ // the possibly incomplete directory invisible; it must not be registered again automatically.
+ if (partitions.isEmpty) {
+ return true
+ }
+
+ val onlyValueInPath =
+ CoreOptions.fromMap(table.options()).formatTablePartitionOnlyValueInPath()
+ // Resolve (and path-safety validate) every partition directory before any mutation, so a
+ // traversal attempt ('.'/'..') fails the whole DROP before unregistering anything.
+ val partitionPaths =
+ partitions.map(spec => resolvePartitionPathWithinTable(orderedSpec(spec), onlyValueInPath))
+ logInfo("Try to drop catalog-registered partitions: " + partitions.mkString(","))
+ requirePartitionManager().dropPartitions(partitions.asJava)
+ val fileIO = table.fileIO()
+ partitionPaths.foreach {
+ partitionPath =>
+ val deleted = fileIO.delete(partitionPath, true)
+ if (!deleted && fileIO.exists(partitionPath)) {
+ throw new java.io.IOException(
+ s"FileIO reported that partition directory $partitionPath was not deleted.")
+ }
+ }
+ true
+ }
+
+ private def validateCatalogRegisteredPartition(
+ partition: JMap[String, String]): JMap[String, String] = {
+ val partitionKeys = table.partitionKeys().asScala
+ if (partitionKeys.exists(key => partition.get(key) == null)) {
+ throw new IllegalStateException(
+ s"Catalog must return a complete partition spec with keys " +
+ s"${partitionKeys.mkString("[", ", ", "]")} for format table " +
+ s"${table.fullName()}, but returned $partition.")
+ }
+
+ val ordered = new util.LinkedHashMap[String, String]()
+ partitionKeys.foreach(key => ordered.put(key, partition.get(key)))
+ ordered
+ }
+
+ private def orderedSpec(spec: JMap[String, String]): util.LinkedHashMap[String, String] = {
+ val ordered = new util.LinkedHashMap[String, String]()
+ table.partitionKeys().asScala.foreach {
+ key => if (spec.containsKey(key)) ordered.put(key, spec.get(key))
+ }
+ ordered
+ }
+
+ /**
+ * Build the partition directory for a spec and verify it stays strictly under the table location.
+ * Value-only path components are validated (including rejecting '.'/'..'), and the normalized
+ * path is checked against the table location so neither DROP (recursive delete) nor ADD (mkdirs)
+ * can escape the table directory via crafted or corrupt partition values.
+ */
+ private def resolvePartitionPathWithinTable(
+ orderedSpec: util.LinkedHashMap[String, String],
+ onlyValueInPath: Boolean): Path = {
+ PartitionPathUtils.validatePartitionSpecForPath(orderedSpec, onlyValueInPath)
+ val tablePath = new Path(table.location())
+ val partitionPath = new Path(
+ tablePath,
+ PartitionPathUtils.generatePartitionPathUtil(orderedSpec, onlyValueInPath)
+ )
+ val normalizedTable = tablePath.toUri.normalize().getPath
+ val tablePrefix = if (normalizedTable.endsWith("/")) normalizedTable else normalizedTable + "/"
+ val normalizedPartition = partitionPath.toUri.normalize().getPath
+ if (!normalizedPartition.startsWith(tablePrefix)) {
+ throw new IllegalArgumentException(
+ s"Resolved partition path $partitionPath escapes the table location $tablePath for " +
+ s"partition spec $orderedSpec of Format Table ${table.fullName()}.")
+ }
+ partitionPath
+ }
+
+ private def requirePartitionManager(): FormatTablePartitionManager = {
+ if (partitionManager == null) {
+ throw new UnsupportedOperationException(
+ s"Catalog-managed partitions are not configured for format table ${table.fullName()}.")
+ }
+ partitionManager
+ }
}
case class PaimonFormatTableWriterBuilder(table: FormatTable, writeSchema: StructType)
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewriteSparkDDLCommands.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewriteSparkDDLCommands.scala
index 5227d1f0f6..5d23412676 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewriteSparkDDLCommands.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/RewriteSparkDDLCommands.scala
@@ -18,14 +18,17 @@
package org.apache.spark.sql.catalyst.parser.extensions
-import org.apache.paimon.spark.catalog.SupportView
-import org.apache.paimon.spark.catalyst.plans.logical.{PaimonDropPartitions, ResolvedIdentifier}
+import org.apache.paimon.spark.SparkTable
+import org.apache.paimon.spark.catalyst.plans.logical.PaimonDropPartitions
+import org.apache.paimon.spark.format.PaimonFormatTable
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.UnresolvedIdentifier
+import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, UnresolvedTable}
import org.apache.spark.sql.catalyst.plans.logical.{DropPartitions, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule
-import org.apache.spark.sql.connector.catalog.{CatalogManager, LookupCatalog}
+import org.apache.spark.sql.connector.catalog.{CatalogManager, LookupCatalog, TableCatalog}
+
+import scala.util.Try
case class RewriteSparkDDLCommands(spark: SparkSession)
extends Rule[LogicalPlan]
@@ -35,10 +38,27 @@ case class RewriteSparkDDLCommands(spark: SparkSession)
override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUp {
- // A new member was added to CreatePaimonView since spark4.0,
- // unapply pattern matching is not used here to ensure compatibility across multiple spark versions.
- case DropPartitions(UnresolvedPaimonRelation(aliasedTable), parts, ifExists, purge) =>
- PaimonDropPartitions(aliasedTable, parts, ifExists, purge)
+ // Rewrite before Spark resolves the partition specs so both Paimon data tables and Format
+ // Tables with catalog-managed partitions can handle partial DROP specs themselves. Format
+ // Tables using filesystem partition discovery are not matched and keep Spark's own resolution
+ // plus the explicit unsupported error at execution. Classification loads the table once through
+ // this rule's own catalog manager (not SparkSession.active), so it resolves against the session
+ // the command actually runs in.
+ case DropPartitions(table, parts, ifExists, purge) if isPaimonDropTarget(table) =>
+ PaimonDropPartitions(table, parts, ifExists, purge)
+ }
+ private def isPaimonDropTarget(plan: LogicalPlan): Boolean = {
+ EliminateSubqueryAliases(plan) match {
+ case UnresolvedTable(CatalogAndIdentifier(catalog: TableCatalog, ident), _, _) =>
+ Try(catalog.loadTable(ident))
+ .map {
+ case _: SparkTable => true
+ case formatTable: PaimonFormatTable => formatTable.hasCatalogManagedPartitions
+ case _ => false
+ }
+ .getOrElse(false)
+ case _ => false
+ }
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/UnresolvedPaimonRelation.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/UnresolvedPaimonRelation.scala
deleted file mode 100644
index a1eaba02a2..0000000000
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/UnresolvedPaimonRelation.scala
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.spark.sql.catalyst.parser.extensions
-
-import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases, UnresolvedTable}
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
-import org.apache.spark.sql.connector.catalog.{PaimonLookupCatalog, TableCatalog}
-
-import scala.util.Try
-
-object UnresolvedPaimonRelation extends PaimonLookupCatalog {
- protected lazy val catalogManager = SparkSession.active.sessionState.catalogManager
-
- def unapply(plan: LogicalPlan): Option[LogicalPlan] = {
- EliminateSubqueryAliases(plan) match {
- case UnresolvedTable(multipartIdentifier, _, _) if isPaimonTable(multipartIdentifier) =>
- Some(plan)
- case _ => None
- }
- }
-
- private def isPaimonTable(multipartIdentifier: Seq[String]): Boolean = {
- multipartIdentifier match {
- case CatalogAndIdentifier(catalog: TableCatalog, ident) =>
- Try(catalog.loadTable(ident))
- .map(t => t.isInstanceOf[org.apache.paimon.spark.SparkTable])
- .getOrElse(false)
- case _ => false
- }
- }
-
-}
diff --git a/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
new file mode 100644
index 0000000000..5bf21ded18
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/java/org/apache/paimon/spark/format/FormatTablePartitionRepairTest.java
@@ -0,0 +1,468 @@
+/*
+ * 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.CoreOptions;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.table.FormatTable;
+import org.apache.paimon.table.format.FormatTablePartitionManager;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for the catalog-managed partition repair engine of Format Tables. */
+class FormatTablePartitionRepairTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void applyDiffsAgainstTheWholeUnfilteredCatalogListing() {
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ catalog.register(Arrays.asList(spec("dt", "20260701"), spec("dt", "20260702")));
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Arrays.asList(
+ spec("dt", "20260701"),
+ spec("dt", "20260702"),
+ spec("dt", "20260703")),
+ Collections.singletonList("dt"),
+ true,
+ false);
+
+ // Only the partition missing from the catalog listing is registered.
+ assertThat(applied).isEqualTo(1);
+ assertThat(catalog.createdPartitions)
+ .containsExactly(Collections.singletonList(spec("dt", "20260703")));
+ // The diff needs every registered partition, so the listing is asked for all of them.
+ assertThat(catalog.requestedPrefixes).containsExactly(Collections.emptyMap());
+ }
+
+ @Test
+ void applyCreatesTheWholeDiffAsAnIdempotentBatch() {
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Arrays.asList(spec("dt", "20260701"), spec("dt", "20260702")),
+ Collections.singletonList("dt"),
+ true,
+ false);
+
+ assertThat(applied).isEqualTo(2);
+ assertThat(catalog.createdPartitions)
+ .containsExactly(Arrays.asList(spec("dt", "20260701"), spec("dt", "20260702")));
+ assertThat(catalog.createIgnoreFlags).containsExactly(true);
+ assertThat(catalog.droppedPartitions).isEmpty();
+ }
+
+ @Test
+ void applyPassesALargeAddDiffToTheCatalogInOneCall() {
+ List<Map<String, String>> filesystemPartitions = new ArrayList<>();
+ for (int index = 0; index < 1001; index++) {
+ filesystemPartitions.add(spec("dt", String.format("%04d", index)));
+ }
+
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ filesystemPartitions,
+ Collections.singletonList("dt"),
+ true,
+ false);
+
+ // Splitting a diff into per-request batches belongs to the partition catalog, so the
+ // repair hands over the complete diff in a single call.
+ assertThat(applied).isEqualTo(1001);
+ assertThat(catalog.createdPartitions).hasSize(1);
+ assertThat(catalog.createdPartitions.get(0)).isEqualTo(filesystemPartitions);
+ assertThat(catalog.createIgnoreFlags).containsExactly(true);
+ }
+
+ @Test
+ void applyPassesALargeDropDiffToTheCatalogInOneCall() {
+ List<Map<String, String>> registered = new ArrayList<>();
+ for (int index = 0; index < 1001; index++) {
+ registered.add(spec("dt", String.format("%04d", index)));
+ }
+
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ catalog.register(registered);
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Collections.emptyList(),
+ Collections.singletonList("dt"),
+ false,
+ true);
+
+ assertThat(applied).isEqualTo(1001);
+ assertThat(catalog.droppedPartitions).hasSize(1);
+ assertThat(catalog.droppedPartitions.get(0)).isEqualTo(registered);
+ assertThat(catalog.createdPartitions).isEmpty();
+ }
+
+ @Test
+ void dropOnlyUnregistersOnlyMissingDirectories() {
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ catalog.register(Arrays.asList(spec("dt", "20260714"), spec("dt", "20260715")));
+
+ // dt=20260715 exists on the filesystem, dt=20260714 does not: only the latter is dropped.
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Collections.singletonList(spec("dt", "20260715")),
+ Collections.singletonList("dt"),
+ false,
+ true);
+
+ assertThat(applied).isEqualTo(1);
+ assertThat(catalog.droppedPartitions)
+ .containsExactly(Collections.singletonList(spec("dt", "20260714")));
+ assertThat(catalog.createdPartitions).isEmpty();
+ }
+
+ @Test
+ void addOnlyNeverDropsStaleCatalogPartitions() {
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ catalog.register(Collections.singletonList(spec("dt", "20260714")));
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Collections.singletonList(spec("dt", "20260715")),
+ Collections.singletonList("dt"),
+ true,
+ false);
+
+ assertThat(applied).isEqualTo(1);
+ assertThat(catalog.createdPartitions)
+ .containsExactly(Collections.singletonList(spec("dt", "20260715")));
+ assertThat(catalog.droppedPartitions).isEmpty();
+ }
+
+ @Test
+ void syncAddsAndDropsInOneCall() {
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ catalog.register(Collections.singletonList(spec("dt", "20260714")));
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Collections.singletonList(spec("dt", "20260715")),
+ Collections.singletonList("dt"),
+ true,
+ true);
+
+ assertThat(applied).isEqualTo(2);
+ assertThat(catalog.createdPartitions)
+ .containsExactly(Collections.singletonList(spec("dt", "20260715")));
+ assertThat(catalog.droppedPartitions)
+ .containsExactly(Collections.singletonList(spec("dt", "20260714")));
+ }
+
+ @Test
+ void applyWithNoDiffDoesNotIssueAnEmptyMutation() {
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ catalog.register(Collections.singletonList(spec("dt", "20260701")));
+
+ int applied =
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Collections.singletonList(spec("dt", "20260701")),
+ Collections.singletonList("dt"),
+ true,
+ true);
+
+ assertThat(applied).isZero();
+ assertThat(catalog.createdPartitions).isEmpty();
+ assertThat(catalog.droppedPartitions).isEmpty();
+ }
+
+ @Test
+ void applyPropagatesFailureAfterAPartiallyAppliedMutation() {
+ IllegalStateException failure =
+ new IllegalStateException("injected partition drop failure");
+ RecordingPartitionManager catalog =
+ new RecordingPartitionManager() {
+ @Override
+ public void dropPartitions(List<Map<String, String>> partitions) {
+ throw failure;
+ }
+ };
+ catalog.register(Collections.singletonList(spec("dt", "20260714")));
+
+ assertThatThrownBy(
+ () ->
+ FormatTablePartitionRepair.apply(
+ catalog,
+ Collections.singletonList(spec("dt", "20260715")),
+ Collections.singletonList("dt"),
+ true,
+ true))
+ .isSameAs(failure);
+ // The ADD half is already committed; a rerun converges from there.
+ assertThat(catalog.createdPartitions)
+ .containsExactly(Collections.singletonList(spec("dt", "20260715")));
+ }
+
+ @Test
+ void repairRegistersRawDirectoryValuesWithoutCastingThem() throws Exception {
+ java.nio.file.Path partitionDirectory =
+ Files.createDirectories(tempDir.resolve("dt=20260701/month=01"));
+ Files.write(
+ partitionDirectory.resolve("data.csv"),
+ Collections.singletonList("1"),
+ StandardCharsets.UTF_8);
+
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ PaimonFormatTable sparkTable =
+ new PaimonFormatTable(
+ monthPartitionedTable(
+ LocalFileIO.create(), tempDir.toUri().toString(), catalog));
+
+ int applied = FormatTablePartitionRepair.repair(sparkTable, true, false);
+
+ // The raw directory value must survive: a scan-based discovery would cast month=01 to 1
+ // and register a spec that no longer round-trips to the real directory.
+ Map<String, String> expected = new LinkedHashMap<>();
+ expected.put("dt", "20260701");
+ expected.put("month", "01");
+ assertThat(applied).isEqualTo(1);
+ assertThat(catalog.createdPartitions).containsExactly(Collections.singletonList(expected));
+ assertThat(catalog.createIgnoreFlags).containsExactly(true);
+ }
+
+ @Test
+ void repairRejectsUnsafeValueOnlyDirectoryBeforeCatalogMutation() throws Exception {
+ Files.createDirectories(tempDir.resolve("%2E%2E"));
+
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ PaimonFormatTable sparkTable =
+ new PaimonFormatTable(formatTable(tempDir.toUri().toString(), true, catalog));
+
+ assertThatThrownBy(() -> FormatTablePartitionRepair.repair(sparkTable, true, true))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("..");
+ assertThat(catalog.requestedPrefixes).isEmpty();
+ assertThat(catalog.createdPartitions).isEmpty();
+ assertThat(catalog.droppedPartitions).isEmpty();
+ }
+
+ @Test
+ void repairFailsClosedWhenTheCatalogListingFails() throws Exception {
+ java.nio.file.Path partitionDirectory =
+ Files.createDirectories(tempDir.resolve("dt=20260715"));
+ Files.write(
+ partitionDirectory.resolve("data.csv"),
+ Collections.singletonList("15"),
+ StandardCharsets.UTF_8);
+
+ IllegalStateException listFailure =
+ new IllegalStateException("injected catalog listing failure");
+ AtomicInteger listCount = new AtomicInteger();
+ RecordingPartitionManager catalog =
+ new RecordingPartitionManager() {
+ @Override
+ public List<Partition> listPartitions(Map<String, String> prefix) {
+ listCount.incrementAndGet();
+ throw listFailure;
+ }
+ };
+ PaimonFormatTable sparkTable =
+ new PaimonFormatTable(formatTable(tempDir.toUri().toString(), catalog));
+
+ assertThatThrownBy(() -> FormatTablePartitionRepair.repair(sparkTable, true, true))
+ .isSameAs(listFailure);
+ assertThat(listCount).hasValue(1);
+ assertThat(catalog.createdPartitions).isEmpty();
+ assertThat(catalog.droppedPartitions).isEmpty();
+ assertThat(partitionDirectory).exists();
+ }
+
+ @Test
+ void repairFailsClosedWhenFilesystemDiscoveryFailsPartway() throws Exception {
+ Files.createDirectories(tempDir.resolve("dt=20260715/month=01"));
+ Files.createDirectories(tempDir.resolve("dt=20260716/month=01"));
+
+ IOException listFailure = new IOException("injected nested filesystem LIST failure");
+ AtomicInteger completedPartitionListings = new AtomicInteger();
+ FileIO fileIO =
+ new LocalFileIO() {
+ @Override
+ public FileStatus[] listStatus(Path path) throws IOException {
+ if ("dt=20260716".equals(path.getName())) {
+ throw listFailure;
+ }
+ FileStatus[] statuses = super.listStatus(path);
+ Arrays.sort(
+ statuses,
+ Comparator.comparing(status -> status.getPath().toString()));
+ if ("dt=20260715".equals(path.getName())) {
+ completedPartitionListings.incrementAndGet();
+ }
+ return statuses;
+ }
+ };
+
+ RecordingPartitionManager catalog = new RecordingPartitionManager();
+ PaimonFormatTable sparkTable =
+ new PaimonFormatTable(
+ monthPartitionedTable(fileIO, tempDir.toUri().toString(), catalog));
+
+ assertThatThrownBy(() -> FormatTablePartitionRepair.repair(sparkTable, true, true))
+ .isInstanceOf(RuntimeException.class)
+ .hasCause(listFailure);
+ // A truncated listing must never reach the diff: no catalog call at all.
+ assertThat(completedPartitionListings).hasValue(1);
+ assertThat(catalog.requestedPrefixes).isEmpty();
+ assertThat(catalog.createdPartitions).isEmpty();
+ assertThat(catalog.droppedPartitions).isEmpty();
+ }
+
+ private static Map<String, String> spec(String key, String value) {
+ Map<String, String> spec = new LinkedHashMap<>();
+ spec.put(key, value);
+ return spec;
+ }
+
+ private static FormatTable formatTable(String location, FormatTablePartitionManager catalog) {
+ return formatTable(location, false, catalog);
+ }
+
+ private static FormatTable formatTable(
+ String location, boolean onlyValueInPath, FormatTablePartitionManager catalog) {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("dt", DataTypes.STRING())
+ .build();
+ return build(
+ LocalFileIO.create(),
+ location,
+ rowType,
+ Collections.singletonList("dt"),
+ onlyValueInPath,
+ catalog);
+ }
+
+ /** A table whose second partition key is an INT, so a cast would rewrite {@code 01} to 1. */
+ private static FormatTable monthPartitionedTable(
+ FileIO fileIO, String location, FormatTablePartitionManager catalog) {
+ RowType rowType =
+ RowType.builder()
+ .field("id", DataTypes.INT())
+ .field("dt", DataTypes.STRING())
+ .field("month", DataTypes.INT())
+ .build();
+ return build(fileIO, location, rowType, Arrays.asList("dt", "month"), false, catalog);
+ }
+
+ private static FormatTable build(
+ FileIO fileIO,
+ String location,
+ RowType rowType,
+ List<String> partitionKeys,
+ boolean onlyValueInPath,
+ FormatTablePartitionManager catalog) {
+ Map<String, String> options = new LinkedHashMap<>();
+ options.put(CoreOptions.METASTORE_PARTITIONED_TABLE.key(), "true");
+ options.put(
+ CoreOptions.FORMAT_TABLE_PARTITION_ONLY_VALUE_IN_PATH.key(),
+ Boolean.toString(onlyValueInPath));
+ return FormatTable.builder()
+ .fileIO(fileIO)
+ .identifier(Identifier.create("db", "t"))
+ .rowType(rowType)
+ .partitionKeys(partitionKeys)
+ .location(new Path(location).toString())
+ .format(FormatTable.Format.CSV)
+ .options(options)
+ .partitionManager(catalog)
+ .build();
+ }
+
+ private static class RecordingPartitionManager implements FormatTablePartitionManager {
+
+ private static final long serialVersionUID = 1L;
+
+ private final List<Map<String, String>> registered = new ArrayList<>();
+ private final List<Map<String, String>> requestedPrefixes = new ArrayList<>();
+ private final List<List<Map<String, String>>> createdPartitions = new ArrayList<>();
+ private final List<Boolean> createIgnoreFlags = new ArrayList<>();
+ private final List<List<Map<String, String>>> droppedPartitions = new ArrayList<>();
+
+ private void register(List<Map<String, String>> partitions) {
+ registered.addAll(partitions);
+ }
+
+ @Override
+ public void createPartitions(List<Map<String, String>> partitions, boolean ignoreIfExists) {
+ createdPartitions.add(new ArrayList<>(partitions));
+ createIgnoreFlags.add(ignoreIfExists);
+ }
+
+ @Override
+ public void dropPartitions(List<Map<String, String>> partitions) {
+ droppedPartitions.add(new ArrayList<>(partitions));
+ }
+
+ @Override
+ public List<Partition> listPartitionsByNames(List<Map<String, String>> partitions) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public List<Partition> listPartitions(Map<String, String> prefix) {
+ requestedPrefixes.add(prefix);
+ List<Partition> partitions = new ArrayList<>(registered.size());
+ for (Map<String, String> spec : registered) {
+ partitions.add(new Partition(spec, 0L, 0L, 0L, 0L, 0, false));
+ }
+ return partitions;
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala
new file mode 100644
index 0000000000..c766a281b9
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/util/OptionUtilsTest.scala
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.util
+
+import org.apache.paimon.CoreOptions.{FORMAT_TABLE_IMPLEMENTATION, METASTORE_PARTITIONED_TABLE}
+import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.fs.local.LocalFileIO
+import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.FormatTable.Format
+import org.apache.paimon.table.format.FormatTablePartitionManager
+import org.apache.paimon.types.{DataTypes, RowType}
+
+import org.apache.spark.sql.internal.SQLConf
+import org.scalatest.funsuite.AnyFunSuite
+
+import java.util.Collections
+
+import scala.collection.JavaConverters._
+
+/** Tests for [[OptionUtils]]. */
+class OptionUtilsTest extends AnyFunSuite {
+
+ test("reject engine implementation from SQL conf for catalog-managed partitions") {
+ val exception = intercept[IllegalArgumentException] {
+ SQLConf.withExistingConf(engineSQLConf) {
+ OptionUtils.copyWithSQLConf(formatTable(withCatalogManagedPartitions = true))
+ }
+ }
+
+ assert(exception.getMessage.contains(METASTORE_PARTITIONED_TABLE.key()))
+ assert(exception.getMessage.contains(FORMAT_TABLE_IMPLEMENTATION.key()))
+ }
+
+ test("allow engine implementation from SQL conf for filesystem-discovered partitions") {
+ val copied = SQLConf.withExistingConf(engineSQLConf) {
+ OptionUtils.copyWithSQLConf(formatTable(withCatalogManagedPartitions = false))
+ }
+
+ assert(copied.options().get(FORMAT_TABLE_IMPLEMENTATION.key()) == "engine")
+ }
+
+ test("reject invalid partition-mode option from SQL conf with option context") {
+ val sqlConf = new SQLConf
+ sqlConf.setConfString(s"spark.paimon.${METASTORE_PARTITIONED_TABLE.key()}", "yes")
+
+ val exception = intercept[IllegalArgumentException] {
+ SQLConf.withExistingConf(sqlConf) {
+ OptionUtils.copyWithSQLConf(formatTable(withCatalogManagedPartitions = false))
+ }
+ }
+
+ assert(exception.getMessage.contains("yes"))
+ assert(exception.getMessage.contains(METASTORE_PARTITIONED_TABLE.key()))
+ }
+
+ test(
+ "conflicting metastore.partitioned-table from SQL conf fails a catalog-managed format table") {
+ val sqlConf = new SQLConf
+ sqlConf.setConfString(s"spark.paimon.${METASTORE_PARTITIONED_TABLE.key()}", "false")
+
+ // Whether a format table's partitions are catalog-managed is a persisted property, not a
+ // dynamic option: a conflicting session value fails the load rather than being silently
+ // dropped. Clear the conflicting Spark config to recover.
+ val exception = intercept[IllegalArgumentException] {
+ SQLConf.withExistingConf(sqlConf) {
+ OptionUtils.copyWithSQLConf(formatTable(withCatalogManagedPartitions = true))
+ }
+ }
+
+ assert(exception.getMessage.contains(METASTORE_PARTITIONED_TABLE.key()))
+ }
+
+ test("conflicting metastore.partitioned-table from SQL conf fails a filesystem format table") {
+ val sqlConf = new SQLConf
+ sqlConf.setConfString(s"spark.paimon.${METASTORE_PARTITIONED_TABLE.key()}", "true")
+
+ val exception = intercept[IllegalArgumentException] {
+ SQLConf.withExistingConf(sqlConf) {
+ OptionUtils.copyWithSQLConf(formatTable(withCatalogManagedPartitions = false))
+ }
+ }
+
+ assert(exception.getMessage.contains(METASTORE_PARTITIONED_TABLE.key()))
+ }
+
+ private def engineSQLConf: SQLConf = {
+ val sqlConf = new SQLConf
+ sqlConf.setConfString(
+ s"spark.paimon.${FORMAT_TABLE_IMPLEMENTATION.key()}",
+ "engine"
+ )
+ sqlConf
+ }
+
+ private def formatTable(withCatalogManagedPartitions: Boolean): FormatTable = {
+ FormatTable
+ .builder()
+ .fileIO(new LocalFileIO)
+ .identifier(Identifier.create("test_db", "format_table"))
+ .rowType(RowType.of(DataTypes.INT(), DataTypes.STRING()))
+ .partitionKeys(Collections.singletonList("dt"))
+ .location("file:///tmp/test_db.db/format_table")
+ .format(Format.PARQUET)
+ .options(Map(
+ METASTORE_PARTITIONED_TABLE.key() -> withCatalogManagedPartitions.toString,
+ FORMAT_TABLE_IMPLEMENTATION.key() -> "paimon").asJava)
+ // A table has catalog-managed partitions when it carries a partition manager, so a fixture
+ // claiming them must supply one.
+ .partitionManager(if (withCatalogManagedPartitions) {
+ FormatTablePartitionManager.create(
+ Identifier.create("test_db", "format_table"),
+ Collections.singletonList("dt"),
+ () => null)
+ } else {
+ null
+ })
+ .build()
+ }
+}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala
new file mode 100644
index 0000000000..7ef72caad0
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/execution/FormatTablePartitionDdlPlanningTest.scala
@@ -0,0 +1,951 @@
+/*
+ * 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.execution
+
+import org.apache.paimon.catalog.{CatalogContext, Identifier}
+import org.apache.paimon.fs.{FileIO, Path}
+import org.apache.paimon.fs.local.LocalFileIO
+import org.apache.paimon.options.Options
+import org.apache.paimon.partition.Partition
+import org.apache.paimon.spark.PaimonSparkTestWithRestCatalogBase
+import org.apache.paimon.spark.catalyst.plans.logical.PaimonDropPartitions
+import org.apache.paimon.spark.format.PaimonFormatTable
+import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.format.FormatTablePartitionManager
+import org.apache.paimon.types.DataTypes
+
+import org.apache.spark.sql.{AnalysisException, Row}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionsException, ResolvedPartitionSpec, ResolvedTable}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, GenericInternalRow}
+import org.apache.spark.sql.catalyst.plans.logical.{AddPartitions, DropPartitions, RepairTable, ShowPartitions}
+import org.apache.spark.sql.connector.catalog.{Identifier => SparkIdentifier, TableCatalog}
+import org.apache.spark.sql.types.StringType
+
+import java.io.IOException
+import java.lang.reflect.InvocationTargetException
+import java.nio.file.Files
+import java.util.{Collections, List => JList, Map => JMap}
+import java.util.concurrent.{Callable, Executors, TimeUnit}
+
+import scala.collection.JavaConverters._
+
+class FormatTablePartitionDdlPlanningTest extends PaimonSparkTestWithRestCatalogBase {
+
+ test("strategy preserves the complete ADD batch for the catalog-managed partition command") {
+ val (table, _) = formatTable(withCatalogManagedPartitions = true)
+ val resolved = ResolvedTable.create(
+ spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog],
+ SparkIdentifier.of(Array("test"), "format_table"),
+ table)
+ val parts = Seq(partition(20260715, 10), partition(20260716, 11))
+
+ val plans = PaimonStrategy(spark).apply(AddPartitions(resolved, parts, ifNotExists = true))
+
+ assert(plans.size == 1)
+ val add = plans.head.asInstanceOf[PaimonAddFormatTablePartitionsExec]
+ assert(add.partSpecs == parts)
+ assert(add.ignoreIfExists)
+ }
+
+ test(
+ "catalog-managed ADD performs no client-side existence lookup and forwards one atomic batch") {
+ val (table, gateway) = formatTable(withCatalogManagedPartitions = true)
+ var refreshCalls = 0
+ val parts = Seq(partition(20260715, 10), partition(20260716, 11))
+ val command = PaimonAddFormatTablePartitionsExec(
+ table,
+ parts,
+ ignoreIfExists = true,
+ () => refreshCalls += 1)
+
+ runCommand(command)
+
+ assert(gateway.createCalls == 1)
+ assert(gateway.lookupCalls == 0)
+ assert(gateway.ignoreIfExists)
+ assert(
+ gateway.created.map(_.asScala.toMap) == Seq(
+ Map("dt" -> "20260715", "hh" -> "10"),
+ Map("dt" -> "20260716", "hh" -> "11")))
+ assert(refreshCalls == 1)
+ }
+
+ test("mock service owns partial repeats, all repeats, atomic failure, and concurrent ADD") {
+ val gateway = new AtomicGateway(Set(Map("dt" -> "20260715", "hh" -> "10")))
+ val table = new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions = true, gateway))
+ val existing = partition(20260715, 10)
+ val added = partition(20260716, 11)
+
+ runCommand(
+ PaimonAddFormatTablePartitionsExec(
+ table,
+ Seq(existing, added),
+ ignoreIfExists = true,
+ () => ()))
+ runCommand(
+ PaimonAddFormatTablePartitionsExec(
+ table,
+ Seq(existing, added),
+ ignoreIfExists = true,
+ () => ()))
+
+ assert(
+ gateway.state == Set(
+ Map("dt" -> "20260715", "hh" -> "10"),
+ Map("dt" -> "20260716", "hh" -> "11")))
+ assert(gateway.batches.take(2).forall(_.size == 2))
+
+ val ordinaryError = intercept[IllegalStateException] {
+ runCommand(
+ PaimonAddFormatTablePartitionsExec(
+ table,
+ Seq(existing, partition(20260717, 12)),
+ ignoreIfExists = false,
+ () => ()))
+ }
+ assert(ordinaryError.getMessage.contains("already exists"))
+ assert(!gateway.state.exists(_("dt") == "20260717"))
+
+ val concurrent = partition(20260718, 13)
+ val executor = Executors.newFixedThreadPool(8)
+ try {
+ val tasks = (1 to 20).map {
+ _ =>
+ executor.submit(new Callable[Unit] {
+ override def call(): Unit = runCommand(
+ PaimonAddFormatTablePartitionsExec(
+ table,
+ Seq(concurrent),
+ ignoreIfExists = true,
+ () => ()))
+ })
+ }
+ tasks.foreach(_.get(30, TimeUnit.SECONDS))
+ } finally {
+ executor.shutdownNow()
+ }
+ assert(gateway.state.count(_("dt") == "20260718") == 1)
+ }
+
+ test(
+ "ADD and DROP always fail with an explicit unsupported error without catalog-managed partitions") {
+ val (table, gateway) = formatTable(withCatalogManagedPartitions = false)
+ val part = partition(20260715, 10)
+
+ val addError = intercept[UnsupportedOperationException] {
+ runCommand(
+ PaimonAddFormatTablePartitionsExec(table, Seq(part), ignoreIfExists = false, () => ()))
+ }
+ assert(addError.getMessage.contains("ADD PARTITION is supported only"))
+ assert(addError.getMessage.contains("catalog-managed"))
+
+ val dropError = intercept[UnsupportedOperationException] {
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(part),
+ ifExists = true,
+ purge = false,
+ () => ()))
+ }
+ assert(dropError.getMessage.contains("DROP PARTITION is supported only"))
+ assert(dropError.getMessage.contains("catalog-managed"))
+ assert(gateway.createCalls == 0)
+ assert(gateway.dropCalls == 0)
+ assert(gateway.lookupCalls == 0)
+ }
+
+ test("SQL ADD PARTITION registers the partition and makes it queryable as empty") {
+ val tableName = "catalog_partition_format_add_sql"
+ withTable(tableName) {
+ sql(s"""CREATE TABLE $tableName (id INT, dt STRING, hh STRING)
+ |USING CSV
+ |PARTITIONED BY (dt, hh)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+
+ sql(s"ALTER TABLE $tableName ADD PARTITION (dt='20260715', hh='10')").collect()
+
+ // The catalog is what makes the partition exist, and the directory is what lets a scan
+ // read it as empty instead of failing.
+ assert(registeredPartitionSpecs(tableName) == Set(Map("dt" -> "20260715", "hh" -> "10")))
+ assert(shownPartitions(tableName) == Set("dt=20260715/hh=10"))
+ checkAnswer(sql(s"SELECT * FROM $tableName"), Seq.empty[Row])
+
+ // Adding it again without IF NOT EXISTS is the catalog's rejection, not a local one.
+ intercept[Exception] {
+ sql(s"ALTER TABLE $tableName ADD PARTITION (dt='20260715', hh='10')").collect()
+ }
+ sql(s"ALTER TABLE $tableName ADD IF NOT EXISTS PARTITION (dt='20260715', hh='10')").collect()
+ assert(registeredPartitionSpecs(tableName) == Set(Map("dt" -> "20260715", "hh" -> "10")))
+ }
+ }
+
+ test("SQL partition DDL on filesystem-discovered partitions always fails with a clear error") {
+ val tableName = "filesystem_format_partition_ddl"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (id INT, dt INT, hh INT) USING CSV " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon') PARTITIONED BY (dt, hh)")
+
+ val addError = intercept[Exception] {
+ sql(s"ALTER TABLE $tableName ADD PARTITION (dt=20260715, hh=10)").collect()
+ }
+ assert(causeMessages(addError).contains("ADD PARTITION is supported only"))
+
+ val dropError = intercept[Exception] {
+ sql(s"ALTER TABLE $tableName DROP IF EXISTS PARTITION (dt=20260715, hh=10)").collect()
+ }
+ assert(causeMessages(dropError).contains("DROP PARTITION is supported only"))
+
+ // Partial specs keep Spark's strict partition-spec resolution when partitions are
+ // discovered from the filesystem.
+ intercept[AnalysisException] {
+ sql(s"ALTER TABLE $tableName DROP PARTITION (dt=20260715)").collect()
+ }
+ }
+ }
+
+ test("rewrite routes catalog-managed DROP PARTITION through PaimonDropPartitions") {
+ val catalogManagedName = "catalog_partition_format_drop_rewrite"
+ val filesystemName = "filesystem_format_drop_rewrite"
+ withTable(catalogManagedName, filesystemName) {
+ sql(s"""CREATE TABLE $catalogManagedName (id INT, dt STRING, hh STRING)
+ |USING CSV
+ |PARTITIONED BY (dt, hh)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+ sql(s"""CREATE TABLE $filesystemName (id INT, dt STRING, hh STRING)
+ |USING CSV
+ |PARTITIONED BY (dt, hh)
+ |TBLPROPERTIES ('format-table.implementation' = 'paimon')
+ |""".stripMargin)
+
+ // RewriteSparkDDLCommands runs inside the extension parser, so parsePlan already returns
+ // the rewritten plan for Format Tables with catalog-managed partitions.
+ val catalogManagedPlan = spark.sessionState.sqlParser.parsePlan(
+ s"ALTER TABLE $catalogManagedName DROP IF EXISTS PARTITION (dt='20260715')")
+ assert(catalogManagedPlan.isInstanceOf[PaimonDropPartitions])
+ val paimonDrop = catalogManagedPlan.asInstanceOf[PaimonDropPartitions]
+ assert(paimonDrop.ifExists)
+ assert(!paimonDrop.purge)
+
+ // Format Tables using filesystem partition discovery keep Spark's own DropPartitions and
+ // strict spec resolution.
+ val filesystemPlan = spark.sessionState.sqlParser.parsePlan(
+ s"ALTER TABLE $filesystemName DROP IF EXISTS PARTITION (dt='20260715')")
+ assert(filesystemPlan.isInstanceOf[DropPartitions])
+ }
+ }
+
+ test("SQL partial DROP with catalog-managed partitions unregisters the matching partitions") {
+ val tableName = "catalog_partition_format_partial_drop_sql"
+ withTable(tableName) {
+ sql(s"""CREATE TABLE $tableName (id INT, dt STRING, hh STRING)
+ |USING CSV
+ |PARTITIONED BY (dt, hh)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+ registerPartitions(
+ tableName,
+ Map("dt" -> "20260715", "hh" -> "10"),
+ Map("dt" -> "20260715", "hh" -> "11"),
+ Map("dt" -> "20260716", "hh" -> "10"))
+
+ // Leading partial spec expands to every registered leaf partition below it.
+ sql(s"ALTER TABLE $tableName DROP PARTITION (dt='20260715')")
+ assert(shownPartitions(tableName) == Set("dt=20260716/hh=10"))
+
+ // Complete specs honor IF EXISTS and fail loudly for unregistered partitions.
+ sql(s"ALTER TABLE $tableName DROP IF EXISTS PARTITION (dt='20990101', hh='00')")
+ val missingError = intercept[Exception] {
+ sql(s"ALTER TABLE $tableName DROP PARTITION (dt='20990101', hh='00')").collect()
+ }
+ assert(
+ Iterator
+ .iterate(missingError: Throwable)(_.getCause)
+ .takeWhile(_ != null)
+ .exists(_.isInstanceOf[NoSuchPartitionsException]))
+ assert(shownPartitions(tableName) == Set("dt=20260716/hh=10"))
+
+ // Non-leading partial specs resolve by partition name through the gateway.
+ sql(s"ALTER TABLE $tableName DROP PARTITION (hh='10')")
+ assert(shownPartitions(tableName) == Set.empty)
+ }
+ }
+
+ test(
+ "strategy preserves the DROP batch and catalog-managed DROP unregisters through the gateway") {
+ val (table, gateway) = formatTable(withCatalogManagedPartitions = true)
+ val resolved = ResolvedTable.create(
+ spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog],
+ SparkIdentifier.of(Array("test"), "format_table"),
+ table)
+ val parts = Seq(partition(20260715, 10))
+ val logical = PaimonDropPartitions(resolved, parts, true, false)
+
+ val plans = PaimonStrategy(spark).apply(logical)
+ assert(plans.size == 1)
+ val command = plans.head.asInstanceOf[PaimonDropFormatTablePartitionsExec]
+ assert(command.partSpecs == parts)
+ assert(command.ifExists)
+ assert(!command.purge)
+
+ var refreshCalls = 0
+ runCommand(command.copy(refreshCache = () => refreshCalls += 1))
+ assert(gateway.lookupCalls == 1)
+ assert(gateway.dropCalls == 1)
+ assert(gateway.dropped.map(_.asScala.toMap) == Seq(Map("dt" -> "20260715", "hh" -> "10")))
+ assert(refreshCalls == 1)
+ }
+
+ test("strategy threads IF EXISTS and PURGE flags into the format-table DROP command") {
+ val (table, _) = formatTable(withCatalogManagedPartitions = true)
+ val resolved = ResolvedTable.create(
+ spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog],
+ SparkIdentifier.of(Array("test"), "format_table"),
+ table)
+ val parts = Seq(partition(20260715, 10))
+
+ val plans =
+ PaimonStrategy(spark).apply(DropPartitions(resolved, parts, ifExists = false, purge = true))
+
+ assert(plans.size == 1)
+ val command = plans.head.asInstanceOf[PaimonDropFormatTablePartitionsExec]
+ assert(command.partSpecs == parts)
+ assert(!command.ifExists)
+ assert(command.purge)
+ }
+
+ test("catalog-managed DROP PARTITION PURGE is rejected before any catalog access") {
+ val (table, gateway) = formatTable(withCatalogManagedPartitions = true)
+
+ val error = intercept[UnsupportedOperationException] {
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(partition(20260715, 10)),
+ ifExists = false,
+ purge = true,
+ () => fail("PURGE must not refresh the cache")))
+ }
+
+ assert(error.getMessage.contains("PURGE"))
+ assert(gateway.lookupCalls == 0)
+ assert(gateway.dropCalls == 0)
+ }
+
+ test("catalog-managed complete DROP without IF EXISTS fails for unregistered partitions") {
+ val fileIO = LocalFileIO.create()
+ val tablePath = new Path(Files.createTempDirectory("format-table-drop-unregistered").toUri)
+ val pendingDir = new Path(tablePath, "dt=20260715/hh=10")
+ var dropCalls = 0
+ var refreshCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = dropCalls += 1
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ throw new AssertionError("Complete specs must resolve through list-by-names")
+ }
+
+ try {
+ fileIO.mkdirs(pendingDir)
+ val table = new PaimonFormatTable(
+ rawFormatTable(withCatalogManagedPartitions = true, gateway, tablePath.toString))
+
+ intercept[NoSuchPartitionsException] {
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(partition(20260715, 10)),
+ ifExists = false,
+ purge = false,
+ () => refreshCalls += 1))
+ }
+
+ // The unregistered directory (e.g. awaiting MSCK registration) must survive untouched.
+ assert(dropCalls == 0)
+ assert(refreshCalls == 0)
+ assert(fileIO.exists(pendingDir))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test(
+ "catalog-managed DROP IF EXISTS drops only registered partitions and preserves pending data") {
+ val fileIO = LocalFileIO.create()
+ val tablePath = new Path(Files.createTempDirectory("format-table-drop-if-exists").toUri)
+ val registeredSpec = Map("dt" -> "20260716", "hh" -> "11")
+ val registeredDir = new Path(tablePath, "dt=20260716/hh=11")
+ val pendingDir = new Path(tablePath, "dt=20260715/hh=10")
+ var lookedUp = Seq.empty[Map[String, String]]
+ var dropped = Seq.empty[Map[String, String]]
+ var refreshCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropped = partitions.asScala.map(_.asScala.toMap).toSeq
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] = {
+ lookedUp = partitions.asScala.map(_.asScala.toMap).toSeq
+ registeredPartitions(
+ partitions.asScala.map(_.asScala.toMap).filter(_ == registeredSpec).toSeq: _*)
+ }
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ throw new AssertionError("Complete specs must resolve through list-by-names")
+ }
+
+ try {
+ fileIO.mkdirs(pendingDir)
+ fileIO.mkdirs(registeredDir)
+ val table = new PaimonFormatTable(
+ rawFormatTable(withCatalogManagedPartitions = true, gateway, tablePath.toString))
+
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(partition(20260715, 10), partition(20260716, 11)),
+ ifExists = true,
+ purge = false,
+ () => refreshCalls += 1))
+
+ assert(
+ lookedUp == Seq(
+ Map("dt" -> "20260715", "hh" -> "10"),
+ Map("dt" -> "20260716", "hh" -> "11")))
+ assert(dropped == Seq(registeredSpec))
+ assert(fileIO.exists(pendingDir))
+ assert(!fileIO.exists(registeredDir))
+ assert(refreshCalls == 1)
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed failed directory deletion stays unregistered and preserves pending data") {
+ val tablePath = new Path(Files.createTempDirectory("format-table-drop-retry-safe").toUri)
+ val failedSpec = Map("dt" -> "20260715", "hh" -> "10")
+ val pendingSpec = Map("dt" -> "20260716", "hh" -> "11")
+ val failedDir = new Path(tablePath, "dt=20260715/hh=10")
+ val pendingDir = new Path(tablePath, "dt=20260716/hh=11")
+ var failFirstDelete = true
+ val fileIO = new LocalFileIO {
+ override def delete(path: Path, recursive: Boolean): Boolean = {
+ if (path == failedDir && failFirstDelete) {
+ failFirstDelete = false
+ throw new IOException("injected exact partition delete failure")
+ }
+ super.delete(path, recursive)
+ }
+ }
+ var registered = Set(failedSpec)
+ var compensationCreates = Seq.empty[(Seq[Map[String, String]], Boolean)]
+ def newGateway(): FormatTablePartitionManager = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ val specs = partitions.asScala.map(_.asScala.toMap).toSeq
+ compensationCreates :+= ((specs, ignoreIfExists))
+ registered ++= specs
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ registered --= partitions.asScala.map(_.asScala.toMap)
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ registeredPartitions(partitions.asScala.map(_.asScala.toMap).filter(registered).toSeq: _*)
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ throw new AssertionError("Exact specs must resolve through list-by-names")
+ }
+ var refreshCalls = 0
+
+ try {
+ fileIO.mkdirs(failedDir)
+ fileIO.mkdirs(pendingDir)
+ val firstTable =
+ new PaimonFormatTable(
+ rawFormatTable(
+ withCatalogManagedPartitions = true,
+ newGateway(),
+ tablePath.toString,
+ fileIO))
+
+ val firstError = intercept[IOException] {
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ firstTable,
+ Seq(partition(20260715, 10)),
+ ifExists = true,
+ purge = false,
+ () => refreshCalls += 1))
+ }
+ assert(firstError.getMessage.contains("injected exact partition delete failure"))
+ // The failed partition stays unregistered and is never recreated automatically.
+ assert(registered.isEmpty)
+ assert(compensationCreates.isEmpty)
+ assert(fileIO.exists(failedDir))
+ assert(fileIO.exists(pendingDir))
+ assert(refreshCalls == 1)
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed ambiguous catalog response does not recreate registration") {
+ val tablePath = new Path(Files.createTempDirectory("format-table-drop-ambiguous").toUri)
+ val droppedSpec = Map("dt" -> "20260715", "hh" -> "10")
+ val droppedDir = new Path(tablePath, "dt=20260715/hh=10")
+ val pendingDir = new Path(tablePath, "dt=20260716/hh=11")
+ val lostResponse = new IOException("injected lost catalog DROP response")
+ var failAfterCommit = true
+ var registered = Set(droppedSpec)
+ var compensationCreates = Seq.empty[(Seq[Map[String, String]], Boolean)]
+ def newGateway(): FormatTablePartitionManager = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ val specs = partitions.asScala.map(_.asScala.toMap).toSeq
+ compensationCreates :+= ((specs, ignoreIfExists))
+ registered ++= specs
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ registered --= partitions.asScala.map(_.asScala.toMap)
+ if (failAfterCommit) {
+ failAfterCommit = false
+ throw lostResponse
+ }
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ registeredPartitions(partitions.asScala.map(_.asScala.toMap).filter(registered).toSeq: _*)
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ throw new AssertionError("Exact specs must resolve through list-by-names")
+ }
+ val firstFileIO = LocalFileIO.create
+ var refreshCalls = 0
+
+ try {
+ firstFileIO.mkdirs(droppedDir)
+ firstFileIO.mkdirs(pendingDir)
+ val firstTable = new PaimonFormatTable(
+ rawFormatTable(
+ withCatalogManagedPartitions = true,
+ newGateway(),
+ tablePath.toString,
+ firstFileIO))
+
+ val firstError = intercept[IOException] {
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ firstTable,
+ Seq(partition(20260715, 10)),
+ ifExists = true,
+ purge = false,
+ () => refreshCalls += 1))
+ }
+
+ assert(firstError eq lostResponse)
+ assert(registered.isEmpty)
+ assert(compensationCreates.isEmpty)
+ assert(firstFileIO.exists(droppedDir))
+ assert(firstFileIO.exists(pendingDir))
+ assert(refreshCalls == 1)
+ } finally {
+ firstFileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed DROP resolves a non-leading partial spec by partition name") {
+ var listedPrefixes = Seq.empty[Map[String, String]]
+ var dropped = Seq.empty[Map[String, String]]
+ val matching = Map("dt" -> "20260715", "hh" -> "10")
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropped = partitions.asScala.map(_.asScala.toMap).toSeq
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] = {
+ listedPrefixes :+= prefix.asScala.toMap
+ registeredPartitions(matching)
+ }
+ }
+ val table = new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions = true, gateway))
+ val hhOnly = ResolvedPartitionSpec(Seq("hh"), new GenericInternalRow(Array[Any](10)))
+
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(hhOnly),
+ ifExists = false,
+ purge = false,
+ () => ()))
+
+ assert(listedPrefixes == Seq(Map.empty))
+ assert(dropped == Seq(matching))
+ }
+
+ test("catalog-managed multiple non-leading partial DROP specs share one catalog traversal") {
+ val first = Map("dt" -> "20260715", "hh" -> "10")
+ val second = Map("dt" -> "20260716", "hh" -> "11")
+ val third = Map("dt" -> "20260717", "hh" -> "10")
+ val unrelated = Map("dt" -> "20260718", "hh" -> "12")
+ var listedPrefixes = Seq.empty[Map[String, String]]
+ var dropped = Seq.empty[Map[String, String]]
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropped = partitions.asScala.map(_.asScala.toMap).toSeq
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] = {
+ listedPrefixes :+= prefix.asScala.toMap
+ registeredPartitions(first, unrelated, second, third)
+ }
+ }
+ val table = new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions = true, gateway))
+ val hhTen = ResolvedPartitionSpec(Seq("hh"), new GenericInternalRow(Array[Any](10)))
+ val hhEleven = ResolvedPartitionSpec(Seq("hh"), new GenericInternalRow(Array[Any](11)))
+
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(hhTen, hhEleven),
+ ifExists = false,
+ purge = false,
+ () => ()))
+
+ // Every partial spec is resolved from a single unfiltered listing.
+ assert(listedPrefixes == Seq(Map.empty))
+ assert(dropped == Seq(first, second, third))
+ assert(dropped.distinct == dropped)
+ }
+
+ test("catalog-managed DROP with an empty batch does not refresh cache") {
+ val (table, gateway) = formatTable(withCatalogManagedPartitions = true)
+ var refreshCalls = 0
+
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq.empty,
+ ifExists = false,
+ purge = false,
+ () => refreshCalls += 1))
+
+ assert(gateway.dropCalls == 0)
+ assert(refreshCalls == 0)
+ }
+
+ test("catalog-managed partial DROP rejects an incomplete catalog result before mutation") {
+ var dropCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = dropCalls += 1
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ registeredPartitions(Map("dt" -> "20260715"))
+ }
+ val table = new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions = true, gateway))
+ val hhOnly = ResolvedPartitionSpec(Seq("hh"), new GenericInternalRow(Array[Any](10)))
+
+ val error = intercept[IllegalStateException] {
+ runCommand(
+ PaimonDropFormatTablePartitionsExec(
+ table,
+ Seq(hhOnly),
+ ifExists = false,
+ purge = false,
+ () => ()))
+ }
+
+ assert(error.getMessage.contains("complete partition spec"))
+ assert(dropCalls == 0)
+ }
+
+ test("MSCK strategy intercepts only catalog-managed partitions and maps the mode flags") {
+ val catalog = spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog]
+ val identifier = SparkIdentifier.of(Array("test"), "format_table")
+ val (catalogManaged, _) = formatTable(withCatalogManagedPartitions = true)
+ val (filesystem, _) = formatTable(withCatalogManagedPartitions = false)
+ val catalogManagedResolved = ResolvedTable.create(catalog, identifier, catalogManaged)
+
+ // (plain MSCK) -> ADD; DROP PARTITIONS -> DROP; SYNC PARTITIONS -> both.
+ Seq((true, false), (false, true), (true, true)).foreach {
+ case (add, drop) =>
+ val plans = PaimonStrategy(spark).apply(RepairTable(catalogManagedResolved, add, drop))
+ assert(plans.size == 1)
+ val repair = plans.head.asInstanceOf[PaimonRepairFormatTablePartitionsExec]
+ assert(repair.addPartitions == add)
+ assert(repair.dropPartitions == drop)
+ }
+
+ // Tables using filesystem partition discovery keep Spark's own v2 rejection: no
+ // interception.
+ val filesystemPlans = PaimonStrategy(spark)
+ .apply(RepairTable(ResolvedTable.create(catalog, identifier, filesystem), true, false))
+ assert(filesystemPlans.isEmpty)
+ }
+
+ test("MSCK repair reuses the sync engine: DROP unregisters catalog-only partitions") {
+ val gateway = new AtomicGateway(Set(Map("dt" -> "20260715", "hh" -> "10")))
+ val table = new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions = true, gateway))
+ var refreshCalls = 0
+
+ // The table location has no partition directories, so the registered partition is
+ // catalog-only; MSCK DROP PARTITIONS must unregister it (metadata-only).
+ runCommand(
+ PaimonRepairFormatTablePartitionsExec(
+ table,
+ addPartitions = false,
+ dropPartitions = true,
+ () => refreshCalls += 1))
+
+ assert(gateway.state.isEmpty)
+ assert(refreshCalls == 1)
+ }
+
+ test("strategy leaves Format Table SHOW PARTITIONS to Spark's default executor") {
+ val output = Seq(AttributeReference("partition", StringType, nullable = false)())
+ val (catalogManaged, _) = formatTable(withCatalogManagedPartitions = true)
+ val (filesystem, _) = formatTable(withCatalogManagedPartitions = false)
+ val catalog = spark.sessionState.catalogManager.currentCatalog.asInstanceOf[TableCatalog]
+ val identifier = SparkIdentifier.of(Array("test"), "format_table")
+ val spec = Some(partition(20260715, 10))
+
+ // SHOW PARTITIONS is served by Spark's default v2 executor via listPartitionIdentifiers
+ // (backed by the catalog listing when partitions are catalog-managed), exactly like every
+ // other Paimon
+ // table, so the strategy must not intercept it.
+ assert(
+ PaimonStrategy(spark)
+ .apply(
+ ShowPartitions(ResolvedTable.create(catalog, identifier, catalogManaged), spec, output))
+ .isEmpty)
+ assert(
+ PaimonStrategy(spark)
+ .apply(ShowPartitions(ResolvedTable.create(catalog, identifier, filesystem), spec, output))
+ .isEmpty)
+ }
+
+ private def partition(dt: Int, hh: Int): ResolvedPartitionSpec =
+ ResolvedPartitionSpec(Seq("dt", "hh"), new GenericInternalRow(Array[Any](dt, hh)))
+
+ /** Partition metadata as a catalog returns it; only the spec matters for these tests. */
+ private def registeredPartitions(specs: Map[String, String]*): JList[Partition] =
+ specs.map(spec => new Partition(spec.asJava, 0L, 0L, 0L, 0L, 0, false)).asJava
+
+ private def formatTable(
+ withCatalogManagedPartitions: Boolean): (PaimonFormatTable, RecordingGateway) = {
+ val gateway = new RecordingGateway
+ (new PaimonFormatTable(rawFormatTable(withCatalogManagedPartitions, gateway)), gateway)
+ }
+
+ private def rawFormatTable(
+ withCatalogManagedPartitions: Boolean,
+ gateway: FormatTablePartitionManager): FormatTable =
+ rawFormatTable(
+ withCatalogManagedPartitions,
+ gateway,
+ new Path(Files.createTempDirectory("format-table-ddl-planning-test").toUri).toString)
+
+ private def rawFormatTable(
+ withCatalogManagedPartitions: Boolean,
+ gateway: FormatTablePartitionManager,
+ location: String): FormatTable = {
+ rawFormatTable(withCatalogManagedPartitions, gateway, location, LocalFileIO.create)
+ }
+
+ private def rawFormatTable(
+ withCatalogManagedPartitions: Boolean,
+ gateway: FormatTablePartitionManager,
+ location: String,
+ fileIO: FileIO): FormatTable = {
+ val options = if (withCatalogManagedPartitions) {
+ Map("metastore.partitioned-table" -> "true")
+ } else {
+ Map.empty[String, String]
+ }
+ FormatTable
+ .builder()
+ .fileIO(fileIO)
+ .identifier(Identifier.create("test", "format_table"))
+ .rowType(
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "dt", DataTypes.INT()),
+ DataTypes.FIELD(2, "hh", DataTypes.INT())))
+ .partitionKeys(Seq("dt", "hh").asJava)
+ .location(location)
+ .format(FormatTable.Format.CSV)
+ .options(options.asJava)
+ .catalogContext(CatalogContext.create(new Options))
+ // A table uses catalog-managed partitions exactly when it carries a partition manager,
+ // so a filesystem-discovered one must not have any.
+ .partitionManager(if (withCatalogManagedPartitions) gateway else null)
+ .build()
+ }
+
+ private def registerPartitions(tableName: String, specs: Map[String, String]*): Unit =
+ paimonCatalog.createPartitions(
+ Identifier.create(dbName0, tableName),
+ specs.map(_.asJava).asJava,
+ true)
+
+ private def registeredPartitionSpecs(tableName: String): Set[Map[String, String]] =
+ paimonCatalog
+ .listPartitions(Identifier.create(dbName0, tableName))
+ .asScala
+ .map(_.spec().asScala.toMap)
+ .toSet
+
+ private def shownPartitions(tableName: String): Set[String] =
+ sql(s"SHOW PARTITIONS $tableName").collect().map(_.getString(0)).toSet
+
+ private def runCommand(command: AnyRef): Unit = {
+ try {
+ command.getClass.getMethod("run").invoke(command)
+ } catch {
+ case e: InvocationTargetException => throw e.getCause
+ }
+ }
+
+ private def causeMessages(error: Throwable): String = {
+ Iterator
+ .iterate(error)(_.getCause)
+ .takeWhile(_ != null)
+ .flatMap(e => Option(e.getMessage))
+ .mkString(" | ")
+ }
+
+ private class RecordingGateway extends FormatTablePartitionManager {
+ var createCalls = 0
+ var lookupCalls = 0
+ var created = Seq.empty[JMap[String, String]]
+ var ignoreIfExists = false
+ var dropCalls = 0
+ var dropped = Seq.empty[JMap[String, String]]
+
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ createCalls += 1
+ created = partitions.asScala.toSeq
+ this.ignoreIfExists = ignoreIfExists
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropCalls += 1
+ dropped = partitions.asScala.toSeq
+ }
+
+ // Reports every requested spec as registered. ADD tests assert lookupCalls == 0 because
+ // Catalog-managed ADD must never perform a client-side existence lookup.
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] = {
+ lookupCalls += 1
+ registeredPartitions(partitions.asScala.map(_.asScala.toMap).toSeq: _*)
+ }
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ private class AtomicGateway(initial: Set[Map[String, String]])
+ extends FormatTablePartitionManager {
+
+ private var partitions = initial
+ var batches = Seq.empty[Seq[Map[String, String]]]
+
+ def state: Set[Map[String, String]] = synchronized(partitions)
+
+ override def createPartitions(
+ partitionsToCreate: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = synchronized {
+ val batch = partitionsToCreate.asScala.map(_.asScala.toMap).toSeq
+ batches :+= batch
+ val duplicates = batch.filter(partitions.contains)
+ if (duplicates.nonEmpty && !ignoreIfExists) {
+ throw new IllegalStateException(s"Partition already exists: ${duplicates.head}")
+ }
+ partitions ++= batch
+ }
+
+ override def dropPartitions(partitionsToDrop: JList[JMap[String, String]]): Unit =
+ synchronized {
+ partitions --= partitionsToDrop.asScala.map(_.asScala.toMap)
+ }
+
+ override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] =
+ throw new AssertionError("ADD must not perform a client-side existence lookup")
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] = synchronized {
+ registeredPartitions(partitions.toSeq: _*)
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionLoadTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionLoadTest.scala
new file mode 100644
index 0000000000..9e39d19565
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/CatalogManagedPartitionLoadTest.scala
@@ -0,0 +1,46 @@
+/*
+ * 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.{PaimonSparkTestWithRestCatalogBase, SparkCatalog}
+
+import org.apache.spark.sql.connector.catalog.{Identifier => SparkIdentifier}
+
+class CatalogManagedPartitionLoadTest extends PaimonSparkTestWithRestCatalogBase {
+
+ test(
+ "a Format Table with catalog-managed partitions loaded through SparkCatalog carries its partition manager") {
+ val tableName = "catalog_partition_format_table_load"
+ withTable(tableName) {
+ sql(
+ s"CREATE TABLE $tableName (id INT, dt STRING) USING CSV " +
+ "TBLPROPERTIES ('format-table.implementation'='paimon', " +
+ "'metastore.partitioned-table'='true') PARTITIONED BY (dt)")
+
+ val sparkCatalog = spark.sessionState.catalogManager
+ .catalog("paimon")
+ .asInstanceOf[SparkCatalog]
+ val sparkTable = sparkCatalog
+ .loadTable(SparkIdentifier.of(Array(dbName0), tableName))
+ .asInstanceOf[PaimonFormatTable]
+
+ assert(sparkTable.partitionManager != null)
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
new file mode 100644
index 0000000000..3cb44485de
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/format/FormatTablePartitionManagementTest.scala
@@ -0,0 +1,768 @@
+/*
+ * 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.catalog.{CatalogContext, Identifier}
+import org.apache.paimon.fs.{FileIO, Path}
+import org.apache.paimon.fs.local.LocalFileIO
+import org.apache.paimon.options.Options
+import org.apache.paimon.partition.Partition
+import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.format.FormatTablePartitionManager
+import org.apache.paimon.types.DataTypes
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.unsafe.types.UTF8String
+
+import java.lang.reflect.{InvocationHandler, InvocationTargetException, Method, Proxy}
+import java.nio.file.Files
+import java.util.{ArrayList, Collections, List => JList, Map => JMap}
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+class FormatTablePartitionManagementTest extends SparkFunSuite {
+
+ test(
+ "catalog-managed ADD forwards the complete batch and ignoreIfExists to the partition catalog") {
+ var forwardedPartitions = Seq.empty[JMap[String, String]]
+ var forwardedIgnoreIfExists = false
+
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ forwardedPartitions = partitions.asScala.toSeq
+ forwardedIgnoreIfExists = ignoreIfExists
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {}
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+ val sparkTable =
+ new PaimonFormatTable(formatTableWithCatalogManagedPartitions(partitionManager = gateway))
+
+ val rows = Array[InternalRow](
+ new GenericInternalRow(Array[Any](20260715, 10)),
+ new GenericInternalRow(Array[Any](20260716, 11)))
+ val properties = Array.fill[JMap[String, String]](2)(Collections.emptyMap())
+ sparkTable.createFormatTablePartitions(rows, properties, ignoreIfExists = true)
+
+ assert(forwardedIgnoreIfExists)
+ assert(
+ forwardedPartitions.map(_.asScala.toMap) == Seq(
+ Map("dt" -> "20260715", "hh" -> "10"),
+ Map("dt" -> "20260716", "hh" -> "11")))
+ }
+
+ test("catalog-managed ADD creates the partition directory") {
+ val fileIO = LocalFileIO.create()
+ val tablePath = new Path(Files.createTempDirectory("catalog-partition-format-add-mkdirs").toUri)
+ val table = formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, emptyGateway)
+ val partitionDir = new Path(tablePath, "dt=20260715/hh=10")
+ try {
+ val sparkTable = new PaimonFormatTable(table)
+ sparkTable.createFormatTablePartitions(
+ Array(partitionRow(20260715, 10)),
+ Array[JMap[String, String]](Collections.emptyMap()),
+ ignoreIfExists = true)
+
+ // ADD PARTITION creates the directory client-side, so a scan sees an empty partition
+ // rather than failing on a missing directory (Hive ADD PARTITION semantics).
+ assert(fileIO.exists(partitionDir))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed DROP rejects a partition value that would escape the table location") {
+ var dropCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = dropCalls += 1
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ val sparkTable =
+ new PaimonFormatTable(
+ stringFormatTableWithCatalogManagedPartitions(gateway, onlyValueInPath = true))
+ val error = intercept[IllegalArgumentException] {
+ sparkTable.dropFormatTablePartitions(
+ Array(Array("dt")),
+ Array(new GenericInternalRow(Array[Any](UTF8String.fromString("..")))))
+ }
+
+ // Traversal values are rejected before any catalog mutation, so no partition is unregistered.
+ assert(error.getMessage.contains(".."))
+ assert(dropCalls == 0)
+ }
+
+ test("catalog-managed ADD with LOCATION is rejected before any catalog RPC") {
+ var createCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = createCalls += 1
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {}
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ val sparkTable =
+ new PaimonFormatTable(formatTableWithCatalogManagedPartitions(partitionManager = gateway))
+ val error = intercept[UnsupportedOperationException] {
+ sparkTable.createFormatTablePartitions(
+ Array(new GenericInternalRow(Array[Any](20260715, 10))),
+ Array(Map("location" -> "file:/tmp/custom").asJava),
+ ignoreIfExists = true)
+ }
+
+ assert(error.getMessage.contains("LOCATION"))
+ assert(createCalls == 0)
+ }
+
+ test("catalog-managed ADD rejects a partition value that would escape the table location") {
+ var createCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = createCalls += 1
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {}
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ val sparkTable =
+ new PaimonFormatTable(
+ stringFormatTableWithCatalogManagedPartitions(gateway, onlyValueInPath = true))
+ val error = intercept[IllegalArgumentException] {
+ sparkTable.createFormatTablePartitions(
+ Array(new GenericInternalRow(Array[Any](UTF8String.fromString("..")))),
+ Array[JMap[String, String]](Collections.emptyMap()),
+ ignoreIfExists = true)
+ }
+
+ // Symmetric with DROP: a traversal value is rejected before any catalog mutation or directory
+ // creation, so the whole ADD fails without registering or creating anything.
+ assert(error.getMessage.contains(".."))
+ assert(createCalls == 0)
+ }
+
+ test("catalog-managed ADD keeps the partition registered when directory creation fails") {
+ val delegate = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-add-mkdirs-failure").toUri)
+ val failure = new java.io.IOException("injected mkdirs failure")
+ val fileIO = throwingMkdirsFileIO(delegate, failure)
+ val gateway = new InMemoryPartitionManager
+ try {
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway))
+ val error = intercept[java.io.IOException] {
+ sparkTable.createFormatTablePartitions(
+ Array(partitionRow(20260715, 10)),
+ Array[JMap[String, String]](Collections.emptyMap()),
+ ignoreIfExists = false)
+ }
+
+ // Registration happens before the client-side mkdirs, so a mkdirs failure surfaces while the
+ // partition stays registered; re-running ADD ... IF NOT EXISTS then creates the directory.
+ assert(error eq failure)
+ assert(gateway.partitions == Seq(Map("dt" -> "20260715", "hh" -> "10")))
+ } finally {
+ delegate.delete(tablePath, true)
+ }
+ }
+
+ test(
+ "catalog-managed DROP PARTITION unregisters first and then deletes the partition directory") {
+ var dropped = Seq.empty[JMap[String, String]]
+ var directoryExistedAtUnregister = false
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-drop-ordering").toUri)
+ val partitionDir = new Path(tablePath, "dt=20260715/hh=10")
+
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ // Ordering contract: the directory must still exist when the catalog unregisters.
+ directoryExistedAtUnregister = fileIO.exists(partitionDir)
+ dropped = partitions.asScala.toSeq
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ try {
+ fileIO.mkdirs(partitionDir)
+ assert(fileIO.exists(partitionDir))
+
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway))
+ assert(
+ sparkTable.dropFormatTablePartitions(
+ Array(Array("dt", "hh")),
+ Array(new GenericInternalRow(Array[Any](20260715, 10)))))
+
+ assert(dropped.map(_.asScala.toMap) == Seq(Map("dt" -> "20260715", "hh" -> "10")))
+ assert(directoryExistedAtUnregister)
+ assert(!fileIO.exists(partitionDir))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed direct partial DROP expands leading values to exact leaf partitions") {
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-direct-partial-drop").toUri)
+ val dtDir = new Path(tablePath, "dt=20260715")
+ val firstDir = new Path(dtDir, "hh=10")
+ val secondDir = new Path(dtDir, "hh=11")
+ val first = partitionSpec(20260715, 10)
+ val second = partitionSpec(20260715, 11)
+ var listedPrefixes = Seq.empty[Map[String, String]]
+ var dropped = Seq.empty[Map[String, String]]
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropped = partitions.asScala.map(_.asScala.toMap).toSeq
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] = {
+ listedPrefixes :+= prefix.asScala.toMap
+ registeredPartitions(first, second)
+ }
+ }
+
+ try {
+ fileIO.mkdirs(firstDir)
+ fileIO.mkdirs(secondDir)
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway))
+
+ sparkTable.dropFormatTablePartitions(
+ Array(Array("dt")),
+ Array(new GenericInternalRow(Array[Any](20260715))))
+
+ // The expansion always runs one unfiltered traversal; constraints apply client-side.
+ assert(listedPrefixes == Seq(Map.empty))
+ assert(dropped == Seq(first, second))
+ assert(!fileIO.exists(firstDir))
+ assert(!fileIO.exists(secondDir))
+ assert(fileIO.exists(dtDir))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed DROP fails when FileIO reports an undeleted partition directory") {
+ val delegate = LocalFileIO.create()
+ val tablePath = new Path(Files.createTempDirectory("catalog-partition-format-drop").toUri)
+ val fileIO = falseDeleteFileIO(delegate)
+ val partitionDir = new Path(tablePath, "dt=20260715/hh=10")
+ delegate.mkdirs(partitionDir)
+ val gateway = new RecordingDropCatalog
+ val table = formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway)
+
+ try {
+ val error = intercept[java.io.IOException] {
+ new PaimonFormatTable(table)
+ .dropFormatTablePartitions(
+ Array(Array("dt", "hh")),
+ Array(new GenericInternalRow(Array[Any](20260715, 10))))
+ }
+
+ assert(error.getMessage.contains("was not deleted"))
+ assert(error.getMessage.contains("dt=20260715"))
+ assert(error.getMessage.contains("hh=10"))
+ assert(gateway.dropCalls == 1)
+ assert(delegate.exists(partitionDir))
+ } finally {
+ delegate.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed DROP treats an already missing partition directory as deleted") {
+ val delegate = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-drop-missing").toUri)
+ val fileIO = falseDeleteFileIO(delegate)
+ val gateway = new RecordingDropCatalog
+ val table = formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway)
+
+ try {
+ assert(
+ new PaimonFormatTable(table)
+ .dropFormatTablePartitions(
+ Array(Array("dt", "hh")),
+ Array(new GenericInternalRow(Array[Any](20260715, 10)))))
+ assert(gateway.dropCalls == 1)
+ } finally {
+ delegate.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed DROP deduplicates overlapping full and partial specs before deleting") {
+ val delegate = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-drop-overlap").toUri)
+ val partitionDir = new Path(tablePath, "dt=20260715/hh=10")
+ var deleteCalls = 0
+ val fileIO = Proxy
+ .newProxyInstance(
+ classOf[FileIO].getClassLoader,
+ Array(classOf[FileIO]),
+ new InvocationHandler {
+ override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = {
+ if (method.getName == "delete" && args(0) == partitionDir) {
+ deleteCalls += 1
+ }
+ try {
+ method.invoke(delegate, Option(args).getOrElse(Array.empty[AnyRef]): _*)
+ } catch {
+ case error: InvocationTargetException => throw error.getCause
+ }
+ }
+ }
+ )
+ .asInstanceOf[FileIO]
+ val matching = partitionSpec(20260715, 10)
+ var dropped = Seq.empty[Map[String, String]]
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropped = partitions.asScala.map(_.asScala.toMap).toSeq
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ registeredPartitions(matching)
+ }
+
+ try {
+ delegate.mkdirs(partitionDir)
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway))
+ sparkTable.dropFormatTablePartitions(
+ Array(Array("dt", "hh"), Array("hh")),
+ Array(partitionRow(20260715, 10), new GenericInternalRow(Array[Any](10))))
+
+ assert(dropped == Seq(matching))
+ assert(deleteCalls == 1)
+ assert(!delegate.exists(partitionDir))
+ assert(delegate.exists(new Path(tablePath, "dt=20260715")))
+ } finally {
+ delegate.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed partial DROP leaves catalog and data untouched when the listing fails") {
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-drop-list-failure").toUri)
+ val partitionDir = new Path(tablePath, "dt=20260715/hh=10")
+ var dropCalls = 0
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = dropCalls += 1
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ throw new TestPartitionListException
+ }
+
+ try {
+ fileIO.mkdirs(partitionDir)
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway))
+
+ intercept[TestPartitionListException] {
+ sparkTable.dropFormatTablePartitions(
+ Array(Array("hh")),
+ Array(new GenericInternalRow(Array[Any](10))))
+ }
+
+ assert(dropCalls == 0)
+ assert(fileIO.exists(partitionDir))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed partial DROP keeps data untouched after an ambiguous catalog response") {
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new Path(Files.createTempDirectory("catalog-partition-format-drop-response-loss").toUri)
+ val firstDir = new Path(tablePath, "dt=20260715/hh=10")
+ val secondDir = new Path(tablePath, "dt=20260715/hh=11")
+ val first = partitionSpec(20260715, 10)
+ val second = partitionSpec(20260715, 11)
+ val registered = mutable.LinkedHashSet(first, second)
+ val responseFailure = new TestCatalogDropResponseException
+ var dropCalls = 0
+ var failDropResponse = true
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit =
+ throw new AssertionError("An ambiguous DROP must not recreate catalog partitions")
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropCalls += 1
+ registered --= partitions.asScala.map(_.asScala.toMap)
+ if (failDropResponse) {
+ failDropResponse = false
+ throw responseFailure
+ }
+ }
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ registeredPartitions(registered.toSeq: _*)
+ }
+
+ try {
+ fileIO.mkdirs(firstDir)
+ fileIO.mkdirs(secondDir)
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString, gateway))
+
+ val error = intercept[TestCatalogDropResponseException] {
+ sparkTable.dropFormatTablePartitions(
+ Array(Array("dt")),
+ Array(new GenericInternalRow(Array[Any](20260715))))
+ }
+
+ assert(error eq responseFailure)
+ assert(dropCalls == 1)
+ assert(registered.isEmpty)
+ assert(fileIO.exists(firstDir))
+ assert(fileIO.exists(secondDir))
+ } finally {
+ fileIO.delete(tablePath, true)
+ }
+ }
+
+ test("catalog-managed ADD forwards each batch whole, without deduplicating across calls") {
+ val first = partitionSpec(20260715, 10)
+ val second = partitionSpec(20260716, 11)
+ val third = partitionSpec(20260717, 12)
+ val gateway = new InMemoryPartitionManager(Seq(first, second))
+ val sparkTable =
+ new PaimonFormatTable(formatTableWithCatalogManagedPartitions(partitionManager = gateway))
+
+ sparkTable.createFormatTablePartitions(
+ Array(partitionRow(20260715, 10), partitionRow(20260716, 11)),
+ Array.fill[JMap[String, String]](2)(Collections.emptyMap()),
+ ignoreIfExists = true)
+ sparkTable.createFormatTablePartitions(
+ Array(partitionRow(20260716, 11), partitionRow(20260717, 12)),
+ Array.fill[JMap[String, String]](2)(Collections.emptyMap()),
+ ignoreIfExists = true)
+
+ // Both batches reach the catalog exactly as asked, including the partition the previous
+ // call already created: resolving duplicates is the catalog's job, not Spark's.
+ assert(gateway.createRequests == Seq((Seq(first, second), true), (Seq(second, third), true)))
+ assert(gateway.lookupCalls == 0)
+ }
+
+ test("catalog-managed strict ADD asks the catalog to reject duplicates, without looking first") {
+ val existing = partitionSpec(20260715, 10)
+ val gateway = new InMemoryPartitionManager(Seq(existing))
+ val sparkTable =
+ new PaimonFormatTable(formatTableWithCatalogManagedPartitions(partitionManager = gateway))
+
+ val error = intercept[TestPartitionAlreadyExistsException] {
+ sparkTable.createFormatTablePartitions(
+ Array(partitionRow(20260715, 10)),
+ Array[JMap[String, String]](Collections.emptyMap()),
+ ignoreIfExists = false)
+ }
+
+ assert(error.partition == existing)
+ // Rejecting the batch is the catalog's decision; Spark must not pre-empt it with a lookup,
+ // which is what would break the batch's atomicity.
+ assert(gateway.lookupCalls == 0)
+ }
+
+ test("catalog-managed ADD preserves typed special partition values") {
+ val gateway = new InMemoryPartitionManager
+ val sparkTable = new PaimonFormatTable(stringFormatTableWithCatalogManagedPartitions(gateway))
+ val values = Seq("a/b", "a=b", "a%", "中文 value")
+ val rows: Seq[InternalRow] = values
+ .map(
+ value => new GenericInternalRow(Array[Any](UTF8String.fromString(value))): InternalRow) :+
+ new GenericInternalRow(Array[Any](null))
+
+ sparkTable.createFormatTablePartitions(
+ rows.toArray,
+ Array.fill[JMap[String, String]](rows.size)(Collections.emptyMap()),
+ ignoreIfExists = true)
+
+ assert(
+ gateway.createRequests == Seq(
+ (values.map(value => Map("dt" -> value)) :+ Map("dt" -> "__DEFAULT_PARTITION__"), true)))
+ }
+
+ private def emptyGateway: FormatTablePartitionManager =
+ new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {}
+
+ override def listPartitionsByNames(
+ partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ private class InMemoryPartitionManager(initialPartitions: Seq[Map[String, String]] = Seq.empty)
+ extends FormatTablePartitionManager {
+
+ private val storedPartitions = mutable.LinkedHashSet(initialPartitions: _*)
+ private var requests = Seq.empty[(Seq[Map[String, String]], Boolean)]
+ var lookupCalls = 0
+
+ def createRequests: Seq[(Seq[Map[String, String]], Boolean)] = synchronized(requests)
+
+ def partitions: Seq[Map[String, String]] = synchronized(storedPartitions.toSeq)
+
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = synchronized {
+ val requested = partitions.asScala.map(_.asScala.toMap).toSeq
+ requests :+= ((requested, ignoreIfExists))
+ if (!ignoreIfExists) {
+ requested.find(storedPartitions.contains).foreach {
+ duplicate => throw new TestPartitionAlreadyExistsException(duplicate)
+ }
+ }
+ storedPartitions ++= requested
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = synchronized {
+ storedPartitions --= partitions.asScala.map(_.asScala.toMap)
+ }
+
+ override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] =
+ synchronized {
+ lookupCalls += 1
+ registeredPartitions(
+ partitions.asScala.map(_.asScala.toMap).filter(storedPartitions.contains).toSeq: _*)
+ }
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ throw new AssertionError("ADD must not enumerate catalog or filesystem partitions")
+ }
+
+ private class TestPartitionAlreadyExistsException(val partition: Map[String, String])
+ extends RuntimeException
+
+ private class TestPartitionListException extends RuntimeException
+
+ private class TestCatalogDropResponseException extends RuntimeException
+
+ private class RecordingDropCatalog extends FormatTablePartitionManager {
+ var dropCalls = 0
+
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {}
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = dropCalls += 1
+
+ override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] =
+ Collections.emptyList()
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] =
+ Collections.emptyList()
+ }
+
+ private def partitionRow(dt: Int, hh: Int): InternalRow =
+ new GenericInternalRow(Array[Any](dt, hh))
+
+ private def partitionSpec(dt: Int, hh: Int): Map[String, String] =
+ Map("dt" -> dt.toString, "hh" -> hh.toString)
+
+ /** Partition metadata as a catalog returns it; only the spec matters for these tests. */
+ private def registeredPartitions(specs: Map[String, String]*): JList[Partition] =
+ specs.map(spec => new Partition(spec.asJava, 0L, 0L, 0L, 0L, 0, false)).asJava
+
+ private def uniqueTableLocation(prefix: String): String =
+ new Path(Files.createTempDirectory(prefix).toUri).toString
+
+ private def formatTableWithCatalogManagedPartitions(
+ fileIO: FileIO = LocalFileIO.create,
+ location: String = uniqueTableLocation("format_table"),
+ partitionManager: FormatTablePartitionManager = null): FormatTable = {
+ FormatTable
+ .builder()
+ .fileIO(fileIO)
+ .identifier(Identifier.create("test_db", "format_table"))
+ .rowType(
+ DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "dt", DataTypes.INT()),
+ DataTypes.FIELD(2, "hh", DataTypes.INT())))
+ .partitionKeys(Seq("dt", "hh").asJava)
+ .location(location)
+ .format(FormatTable.Format.CSV)
+ .options(Map("metastore.partitioned-table" -> "true").asJava)
+ .catalogContext(CatalogContext.create(new Options))
+ .partitionManager(partitionManager)
+ .build()
+ }
+
+ private def throwingMkdirsFileIO(delegate: FileIO, failure: Throwable): FileIO = {
+ Proxy
+ .newProxyInstance(
+ classOf[FileIO].getClassLoader,
+ Array(classOf[FileIO]),
+ new InvocationHandler {
+ override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = {
+ if (method.getName == "mkdirs") {
+ throw failure
+ } else {
+ try {
+ method.invoke(delegate, Option(args).getOrElse(Array.empty[AnyRef]): _*)
+ } catch {
+ case error: InvocationTargetException => throw error.getCause
+ }
+ }
+ }
+ }
+ )
+ .asInstanceOf[FileIO]
+ }
+
+ private def falseDeleteFileIO(delegate: FileIO): FileIO = {
+ Proxy
+ .newProxyInstance(
+ classOf[FileIO].getClassLoader,
+ Array(classOf[FileIO]),
+ new InvocationHandler {
+ override def invoke(proxy: Any, method: Method, args: Array[AnyRef]): AnyRef = {
+ if (method.getName == "delete" && method.getParameterCount == 2) {
+ java.lang.Boolean.FALSE
+ } else {
+ try {
+ method.invoke(delegate, Option(args).getOrElse(Array.empty[AnyRef]): _*)
+ } catch {
+ case error: InvocationTargetException => throw error.getCause
+ }
+ }
+ }
+ }
+ )
+ .asInstanceOf[FileIO]
+ }
+
+ private def stringFormatTableWithCatalogManagedPartitions(
+ partitionManager: FormatTablePartitionManager,
+ onlyValueInPath: Boolean = false): FormatTable = {
+ FormatTable
+ .builder()
+ .fileIO(LocalFileIO.create)
+ .identifier(Identifier.create("test_db", "format_table_special"))
+ .rowType(DataTypes.ROW(
+ DataTypes.FIELD(0, "id", DataTypes.INT()),
+ DataTypes.FIELD(1, "dt", DataTypes.STRING())))
+ .partitionKeys(Seq("dt").asJava)
+ .location(uniqueTableLocation("format_table_special"))
+ .format(FormatTable.Format.CSV)
+ .options(Map(
+ "metastore.partitioned-table" -> "true",
+ "format-table.partition-path-only-value" -> onlyValueInPath.toString).asJava)
+ .catalogContext(CatalogContext.create(new Options))
+ .partitionManager(partitionManager)
+ .build()
+ }
+}
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
new file mode 100644
index 0000000000..3997341b45
--- /dev/null
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionMsckRepairTest.scala
@@ -0,0 +1,669 @@
+/*
+ * 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.sql
+
+import org.apache.paimon.catalog.Identifier
+import org.apache.paimon.fs.Path
+import org.apache.paimon.partition.Partition
+import org.apache.paimon.spark.{PaimonSparkTestWithRestCatalogBase, SparkCatalog}
+import org.apache.paimon.spark.execution.PaimonRepairFormatTablePartitionsExec
+import org.apache.paimon.spark.format.PaimonFormatTable
+import org.apache.paimon.table.FormatTable
+import org.apache.paimon.table.format.FormatTablePartitionManager
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.connector.catalog.{Identifier => SparkIdentifier, Table => SparkTable}
+import org.apache.spark.sql.execution.CommandResultExec
+
+import java.lang.reflect.InvocationTargetException
+import java.util.{List => JList, Map => JMap}
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable
+
+class CatalogManagedPartitionMsckRepairTest extends PaimonSparkTestWithRestCatalogBase {
+
+ override protected def sparkConf: SparkConf =
+ super.sparkConf
+ .set("spark.sql.catalog.paimon", classOf[FaultInjectingFormatTableSparkCatalog].getName)
+
+ override protected def beforeEach(): Unit = {
+ MsckFaultInjection.reset()
+ super.beforeEach()
+ }
+
+ override protected def afterEach(): Unit = {
+ try {
+ super.afterEach()
+ } finally {
+ MsckFaultInjection.reset()
+ }
+ }
+
+ test("bare MSCK, explicit ADD, and REPAIR alias register filesystem-only partitions") {
+ val tableName = "msck_add_forms"
+ val first = "20260715"
+ val second = "20260716"
+ val third = "20260717"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+
+ writeCsvPartition(tableName, first, 15, "bare-msck")
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName")
+ assertPartitionState(tableName, Set(first))
+
+ writeCsvPartition(tableName, second, 16, "explicit-add")
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName ADD PARTITIONS")
+ assertPartitionState(tableName, Set(first, second))
+
+ writeCsvPartition(tableName, third, 17, "repair-alias")
+ executeCatalogManagedRepair(s"REPAIR TABLE paimon.$dbName0.$tableName ADD PARTITIONS")
+ assertPartitionState(tableName, Set(first, second, third))
+
+ executeCatalogManagedRepair(s"REPAIR TABLE paimon.$dbName0.$tableName ADD PARTITIONS")
+
+ assertPartitionState(tableName, Set(first, second, third))
+ assert(MsckFaultInjection.createCalls == 3)
+ assertRowsAndChecksum(
+ tableName,
+ Seq(
+ Row(15, "bare-msck", first),
+ Row(16, "explicit-add", second),
+ Row(17, "repair-alias", third)),
+ 48L)
+ }
+ }
+
+ test("DROP unregisters only catalog-only partitions and preserves valid files") {
+ val tableName = "msck_drop_catalog_only"
+ val catalogOnly = "20260714"
+ val common = "20260715"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ val commonPath = writeCsvPartition(tableName, common, 15, "common")
+ registerPartitions(tableName, catalogOnly, common)
+
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName DROP PARTITIONS")
+
+ assertPartitionState(tableName, Set(common))
+ assert(formatTable(tableName).fileIO().exists(commonPath))
+ assertRowsAndChecksum(tableName, Seq(Row(15, "common", common)), 15L)
+
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName DROP PARTITIONS")
+ assertPartitionState(tableName, Set(common))
+ assert(MsckFaultInjection.createCalls == 0)
+ assert(MsckFaultInjection.dropCalls == 1)
+ }
+ }
+
+ test("SYNC converges both directions and is idempotent") {
+ val tableName = "msck_sync_both_directions"
+ val catalogOnly = "20260714"
+ val filesystemOnly = "20260715"
+ val common = "20260716"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ val filesystemPath =
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ val commonPath = writeCsvPartition(tableName, common, 16, "common")
+ registerPartitions(tableName, catalogOnly, common)
+
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS")
+
+ assertPartitionState(tableName, Set(filesystemOnly, common))
+ assert(formatTable(tableName).fileIO().exists(filesystemPath))
+ assert(formatTable(tableName).fileIO().exists(commonPath))
+ assertRowsAndChecksum(
+ tableName,
+ Seq(Row(15, "filesystem-only", filesystemOnly), Row(16, "common", common)),
+ 31L)
+ assert(MsckFaultInjection.createCalls == 1)
+ assert(MsckFaultInjection.dropCalls == 1)
+
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS")
+ assertPartitionState(tableName, Set(filesystemOnly, common))
+ assert(MsckFaultInjection.createCalls == 1)
+ assert(MsckFaultInjection.dropCalls == 1)
+ }
+ }
+
+ test("MSCK rejects filesystem-discovered Format Tables and native Paimon tables") {
+ val filesystemPartitioned = "msck_filesystem_format_table"
+ val native = "msck_native_paimon_table"
+ val partition = "20260715"
+
+ withTable(filesystemPartitioned, native) {
+ createFormatTableWithFilesystemPartitions(filesystemPartitioned)
+ writeCsvPartition(filesystemPartitioned, partition, 15, "filesystem")
+ checkAnswer(
+ sql(s"SELECT id, payload, dt FROM $filesystemPartitioned"),
+ Seq(Row(15, "filesystem", partition)))
+
+ assertMsckRejected(s"MSCK REPAIR TABLE paimon.$dbName0.$filesystemPartitioned")
+ checkAnswer(
+ sql(s"SELECT id, payload, dt FROM $filesystemPartitioned"),
+ Seq(Row(15, "filesystem", partition)))
+
+ sql(
+ s"CREATE TABLE $native (id INT, payload STRING, dt STRING) " +
+ "USING paimon PARTITIONED BY (dt)")
+ sql(s"INSERT INTO $native VALUES (16, 'native', '$partition')")
+
+ assertMsckRejected(s"MSCK REPAIR TABLE paimon.$dbName0.$native")
+ checkAnswer(sql(s"SELECT id, payload, dt FROM $native"), Seq(Row(16, "native", partition)))
+ assert(MsckFaultInjection.createCalls == 0)
+ assert(MsckFaultInjection.dropCalls == 0)
+ }
+ }
+
+ test("SYNC propagates catalog LIST failure without mutating either side") {
+ val tableName = "msck_sync_list_failure"
+ val catalogOnly = "20260714"
+ val filesystemOnly = "20260715"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ val filesystemPath =
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ registerPartitions(tableName, catalogOnly)
+ MsckFaultInjection.failList = true
+
+ val error = intercept[Exception] {
+ sql(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS").collect()
+ }
+
+ assert(causeMessages(error).contains(MsckFaultInjection.LIST_FAILURE))
+ assert(MsckFaultInjection.listCalls == 1)
+ assert(MsckFaultInjection.createCalls == 0)
+ assert(MsckFaultInjection.dropCalls == 0)
+ assert(registeredPartitions(tableName) == Set(catalogOnly))
+ assert(formatTable(tableName).fileIO().exists(filesystemPath))
+
+ MsckFaultInjection.failList = false
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS")
+ assertPartitionState(tableName, Set(filesystemOnly))
+ assertRowsAndChecksum(tableName, Seq(Row(15, "filesystem-only", filesystemOnly)), 15L)
+ }
+ }
+
+ test("direct repair mutates nothing but still refreshes when catalog LIST fails") {
+ val tableName = "msck_direct_list_failure"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ val gateway = new StatefulFaultCatalog
+ gateway.failList = true
+ var refreshCalls = 0
+ val command = PaimonRepairFormatTablePartitionsExec(
+ sparkTable(tableName, gateway),
+ addPartitions = true,
+ dropPartitions = true,
+ () => refreshCalls += 1)
+
+ val error = intercept[IllegalStateException] {
+ runCommand(command)
+ }
+
+ assert(error.getMessage == MsckFaultInjection.LIST_FAILURE)
+ assert(gateway.createCalls == 0)
+ assert(gateway.dropCalls == 0)
+ // A failed response cannot prove the service did nothing, so every repair attempt
+ // refreshes the Spark-side caches.
+ assert(refreshCalls == 1)
+ }
+ }
+
+ test("direct repair refreshes after create applies and loses its response") {
+ val tableName = "msck_direct_create_lost_response"
+ val filesystemOnly = "20260715"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ val gateway = new StatefulFaultCatalog
+ gateway.failCreateAfterApply = true
+ var refreshCalls = 0
+ val command = PaimonRepairFormatTablePartitionsExec(
+ sparkTable(tableName, gateway),
+ addPartitions = true,
+ dropPartitions = true,
+ () => refreshCalls += 1)
+
+ val error = intercept[IllegalStateException] {
+ runCommand(command)
+ }
+
+ assert(error.getMessage == MsckFaultInjection.CREATE_LOST_RESPONSE)
+ assert(gateway.partitions == Set(Map("dt" -> filesystemOnly)))
+ assert(gateway.createCalls == 1)
+ assert(gateway.dropCalls == 0)
+ assert(refreshCalls == 1)
+ }
+ }
+
+ test("direct repair keeps the DROP failure primary when refresh also fails") {
+ val tableName = "msck_direct_drop_and_refresh_failure"
+ val catalogOnly = "20260714"
+ val filesystemOnly = "20260715"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ val gateway = new StatefulFaultCatalog(Set(Map("dt" -> catalogOnly)))
+ gateway.failDrop = true
+ var refreshCalls = 0
+ val command = PaimonRepairFormatTablePartitionsExec(
+ sparkTable(tableName, gateway),
+ addPartitions = true,
+ dropPartitions = true,
+ () => {
+ refreshCalls += 1
+ throw new IllegalStateException(MsckFaultInjection.REFRESH_FAILURE)
+ }
+ )
+
+ val error = intercept[IllegalStateException] {
+ runCommand(command)
+ }
+
+ // The DROP failure is what explains the repair, so it propagates and the refresh
+ // failure rides along; the ADD stays durable and the refresh is still attempted.
+ assert(error.getMessage == MsckFaultInjection.DROP_FAILURE)
+ assert(error.getSuppressed.map(_.getMessage).toSeq == Seq(MsckFaultInjection.REFRESH_FAILURE))
+ assert(gateway.partitions == Set(Map("dt" -> catalogOnly), Map("dt" -> filesystemOnly)))
+ assert(gateway.createCalls == 1)
+ assert(gateway.dropCalls == 1)
+ assert(refreshCalls == 1)
+ }
+ }
+
+ test("direct repair does not suppress a DROP throwable into itself") {
+ val tableName = "msck_direct_same_drop_refresh_failure"
+ val catalogOnly = "20260714"
+ val filesystemOnly = "20260715"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ val gateway = new StatefulFaultCatalog(Set(Map("dt" -> catalogOnly)))
+ val dropFailure = new IllegalStateException(MsckFaultInjection.DROP_FAILURE)
+ gateway.dropFailure = dropFailure
+ var refreshCalls = 0
+ val command = PaimonRepairFormatTablePartitionsExec(
+ sparkTable(tableName, gateway),
+ addPartitions = true,
+ dropPartitions = true,
+ () => {
+ refreshCalls += 1
+ throw dropFailure
+ }
+ )
+
+ val error = intercept[Throwable] {
+ runCommand(command)
+ }
+
+ // Refresh rethrows the very throwable the DROP failed with. Attaching it to itself would
+ // throw IllegalArgumentException, so the two must be recognised as the same failure.
+ assert(error eq dropFailure)
+ assert(!error.isInstanceOf[MatchError])
+ assert(error.getSuppressed.isEmpty)
+ assert(gateway.partitions == Set(Map("dt" -> catalogOnly), Map("dt" -> filesystemOnly)))
+ assert(gateway.createCalls == 1)
+ assert(gateway.dropCalls == 1)
+ assert(refreshCalls == 1)
+ }
+ }
+
+ test("direct repair surfaces the refresh failure when the operation succeeds") {
+ val tableName = "msck_direct_success_refresh_failure"
+ val filesystemOnly = "20260715"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ val gateway = new StatefulFaultCatalog
+ var refreshCalls = 0
+ val command = PaimonRepairFormatTablePartitionsExec(
+ sparkTable(tableName, gateway),
+ addPartitions = true,
+ dropPartitions = false,
+ () => {
+ refreshCalls += 1
+ throw new IllegalStateException(MsckFaultInjection.REFRESH_FAILURE)
+ }
+ )
+
+ val error = intercept[IllegalStateException] {
+ runCommand(command)
+ }
+
+ // With no operation failure to explain the repair, the refresh failure is the primary error
+ // (nothing to suppress it into); the ADD stays durable and refresh is attempted once.
+ assert(error.getMessage == MsckFaultInjection.REFRESH_FAILURE)
+ assert(error.getSuppressed.isEmpty)
+ assert(gateway.partitions == Set(Map("dt" -> filesystemOnly)))
+ assert(gateway.createCalls == 1)
+ assert(gateway.dropCalls == 0)
+ assert(refreshCalls == 1)
+ }
+ }
+
+ test("SYNC invalidates cached data and converges after ADD succeeds but DROP fails") {
+ val tableName = "msck_partial_sync_failure"
+ val filesystemOnly = "20260715"
+ val catalogOnly = "20260714"
+
+ withTable(tableName) {
+ createFormatTableWithCatalogManagedPartitions(tableName)
+ sql(s"CACHE TABLE $tableName").collect()
+ assert(spark.catalog.isCached(tableName))
+ checkAnswer(sql(s"SELECT * FROM $tableName"), Seq.empty[Row])
+
+ writeCsvPartition(tableName, filesystemOnly, 15, "filesystem-only")
+ registerPartitions(tableName, catalogOnly)
+ MsckFaultInjection.failDrop = true
+
+ val error = intercept[Exception] {
+ sql(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS").collect()
+ }
+
+ assert(causeMessages(error).contains(MsckFaultInjection.DROP_FAILURE))
+ assert(MsckFaultInjection.createCalls == 1)
+ assert(MsckFaultInjection.dropCalls == 1)
+ assert(registeredPartitions(tableName) == Set(filesystemOnly, catalogOnly))
+ assert(spark.catalog.isCached(tableName))
+
+ // ADD is already durable even though the command failed. A stale CACHE TABLE must not hide it.
+ checkAnswer(
+ sql(s"SELECT id, payload, dt FROM $tableName WHERE dt = '$filesystemOnly'"),
+ Seq(Row(15, "filesystem-only", filesystemOnly)))
+
+ MsckFaultInjection.failDrop = false
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS")
+
+ assertPartitionState(tableName, Set(filesystemOnly))
+ assert(MsckFaultInjection.createCalls == 1)
+ assert(MsckFaultInjection.dropCalls == 2)
+ assertRowsAndChecksum(tableName, Seq(Row(15, "filesystem-only", filesystemOnly)), 15L)
+
+ executeCatalogManagedRepair(s"MSCK REPAIR TABLE paimon.$dbName0.$tableName SYNC PARTITIONS")
+ assertPartitionState(tableName, Set(filesystemOnly))
+ assert(MsckFaultInjection.createCalls == 1)
+ assert(MsckFaultInjection.dropCalls == 2)
+ }
+ }
+
+ private def createFormatTableWithCatalogManagedPartitions(tableName: String): Unit =
+ sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING)
+ |USING CSV
+ |PARTITIONED BY (dt)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'true')
+ |""".stripMargin)
+
+ private def createFormatTableWithFilesystemPartitions(tableName: String): Unit =
+ sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING)
+ |USING CSV
+ |PARTITIONED BY (dt)
+ |TBLPROPERTIES (
+ | 'format-table.implementation' = 'paimon',
+ | 'metastore.partitioned-table' = 'false')
+ |""".stripMargin)
+
+ private def executeCatalogManagedRepair(statement: String): Unit = {
+ val result = sql(statement)
+ val resultPlan = result.queryExecution.executedPlan
+ assert(resultPlan.isInstanceOf[CommandResultExec], resultPlan.treeString)
+ val commandPlan = resultPlan.asInstanceOf[CommandResultExec].commandPhysicalPlan
+ assert(commandPlan.isInstanceOf[PaimonRepairFormatTablePartitionsExec], commandPlan.treeString)
+ result.collect()
+ }
+
+ private def assertMsckRejected(statement: String): Unit = {
+ val error = intercept[Exception] {
+ sql(statement).collect()
+ }
+ val messages = causeMessages(error)
+ assert(messages.contains("MSCK REPAIR TABLE"), messages)
+ assert(messages.toLowerCase(java.util.Locale.ROOT).contains("not supported"), messages)
+ }
+
+ private def assertPartitionState(tableName: String, expected: Set[String]): Unit = {
+ assert(registeredPartitions(tableName) == expected)
+ assert(
+ sql(s"SHOW PARTITIONS $tableName").collect().map(_.getString(0)).toSet ==
+ expected.map(value => s"dt=$value"))
+ }
+
+ private def assertRowsAndChecksum(
+ tableName: String,
+ expectedRows: Seq[Row],
+ expectedIdSum: Long): Unit = {
+ checkAnswer(sql(s"SELECT id, payload, dt FROM $tableName ORDER BY id"), expectedRows)
+ checkAnswer(
+ sql(s"SELECT COUNT(*), COALESCE(SUM(id), 0) FROM $tableName"),
+ Seq(Row(expectedRows.size.toLong, expectedIdSum)))
+ }
+
+ private def runCommand(command: AnyRef): Unit = {
+ try {
+ command.getClass.getMethod("run").invoke(command)
+ } catch {
+ case error: InvocationTargetException => throw error.getCause
+ }
+ }
+
+ private def tableIdentifier(tableName: String): Identifier =
+ Identifier.create(dbName0, tableName)
+
+ private def formatTable(tableName: String): FormatTable =
+ paimonCatalog.getTable(tableIdentifier(tableName)).asInstanceOf[FormatTable]
+
+ private def sparkTable(
+ tableName: String,
+ partitionManager: FormatTablePartitionManager): PaimonFormatTable =
+ new PaimonFormatTable(
+ MsckTestFixtures.withPartitionManager(formatTable(tableName), partitionManager))
+
+ private def writeCsvPartition(
+ tableName: String,
+ partition: String,
+ id: Int,
+ payload: String): Path = {
+ val table = formatTable(tableName)
+ val partitionPath = new Path(table.location(), s"dt=$partition")
+ table.fileIO().mkdirs(partitionPath)
+ table.fileIO().writeFile(new Path(partitionPath, f"part-$id%05d.csv"), s"$id,$payload\n", false)
+ partitionPath
+ }
+
+ private def registerPartitions(tableName: String, partitions: String*): Unit =
+ paimonCatalog.createPartitions(
+ tableIdentifier(tableName),
+ partitions.map(value => Map("dt" -> value).asJava).asJava,
+ true)
+
+ private def registeredPartitions(tableName: String): Set[String] =
+ paimonCatalog
+ .listPartitions(tableIdentifier(tableName))
+ .asScala
+ .map(_.spec().get("dt"))
+ .toSet
+
+ private def causeMessages(error: Throwable): String =
+ Iterator
+ .iterate(error)(_.getCause)
+ .takeWhile(_ != null)
+ .flatMap(cause => Option(cause.getMessage))
+ .mkString(" | ")
+}
+
+private[sql] object MsckFaultInjection {
+ val DROP_FAILURE = "injected MSCK partition drop failure"
+ val LIST_FAILURE = "injected MSCK partition list failure"
+ val CREATE_LOST_RESPONSE = "injected MSCK create lost response"
+ val REFRESH_FAILURE = "injected MSCK cache refresh failure"
+
+ @volatile var failDrop = false
+ @volatile var failList = false
+ @volatile var createCalls = 0
+ @volatile var dropCalls = 0
+ @volatile var listCalls = 0
+
+ def reset(): Unit = {
+ failDrop = false
+ failList = false
+ createCalls = 0
+ dropCalls = 0
+ listCalls = 0
+ }
+}
+
+private[sql] class StatefulFaultCatalog(initial: Set[Map[String, String]] = Set.empty)
+ extends FormatTablePartitionManager {
+
+ private val state = mutable.LinkedHashSet(initial.toSeq: _*)
+ var failList = false
+ var failCreateAfterApply = false
+ var failDrop = false
+ var dropFailure: RuntimeException = null
+ var createCalls = 0
+ var dropCalls = 0
+
+ def partitions: Set[Map[String, String]] = state.toSet
+
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ createCalls += 1
+ state ++= partitions.asScala.map(_.asScala.toMap)
+ if (failCreateAfterApply) {
+ throw new IllegalStateException(MsckFaultInjection.CREATE_LOST_RESPONSE)
+ }
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ dropCalls += 1
+ if (dropFailure != null) {
+ throw dropFailure
+ }
+ if (failDrop) {
+ throw new IllegalStateException(MsckFaultInjection.DROP_FAILURE)
+ }
+ state --= partitions.asScala.map(_.asScala.toMap)
+ }
+
+ override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] =
+ MsckTestFixtures.toPartitions(
+ partitions.asScala.map(_.asScala.toMap).filter(state.contains).toSeq)
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] = {
+ if (failList) {
+ throw new IllegalStateException(MsckFaultInjection.LIST_FAILURE)
+ }
+ MsckTestFixtures.toPartitions(state.toSeq)
+ }
+}
+
+/** Test helpers shared by the fault-injecting catalogs of this suite. */
+private[sql] object MsckTestFixtures {
+
+ /** Partition metadata as a catalog returns it; only the spec matters for these tests. */
+ def toPartitions(specs: Seq[Map[String, String]]): JList[Partition] =
+ specs.map(spec => new Partition(spec.asJava, 0L, 0L, 0L, 0L, 0, false)).asJava
+
+ private val rebound: JMap[FormatTable, FormatTable] =
+ java.util.Collections.synchronizedMap(new java.util.IdentityHashMap[FormatTable, FormatTable]())
+
+ /** Rebind a table once per source instance, so repeated loads return the same table. */
+ def rebindOnce(table: FormatTable, partitionManager: FormatTablePartitionManager): FormatTable =
+ rebound.computeIfAbsent(table, _ => withPartitionManager(table, partitionManager))
+
+ /** Rebind a Format Table to another partition catalog, as loading it from a catalog would. */
+ def withPartitionManager(
+ table: FormatTable,
+ partitionManager: FormatTablePartitionManager): FormatTable =
+ FormatTable
+ .builder()
+ .fileIO(table.fileIO())
+ .identifier(Identifier.fromString(table.fullName()))
+ .rowType(table.rowType())
+ .partitionKeys(table.partitionKeys())
+ .location(table.location())
+ .format(table.format())
+ .options(table.options())
+ .comment(table.comment().orElse(null))
+ .catalogContext(table.catalogContext())
+ .partitionManager(partitionManager)
+ .build()
+}
+
+/** Spark-loadable test catalog which delegates all state to the embedded Paimon REST catalog. */
+private[sql] class FaultInjectingFormatTableSparkCatalog extends SparkCatalog {
+
+ override def loadTable(ident: SparkIdentifier): SparkTable = {
+ super.loadTable(ident) match {
+ case table: PaimonFormatTable if table.partitionManager != null =>
+ // The catalog hands out one table instance per table, and Spark's cached plans compare
+ // that instance; rebinding must be memoized per source instance or CACHE TABLE stops
+ // matching the relation it cached.
+ new PaimonFormatTable(
+ MsckTestFixtures.rebindOnce(
+ table.table,
+ new FaultInjectingFormatTablePartitionManager(table.partitionManager)))
+ case table => table
+ }
+ }
+}
+
+private[sql] class FaultInjectingFormatTablePartitionManager(delegate: FormatTablePartitionManager)
+ extends FormatTablePartitionManager {
+
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean): Unit = {
+ delegate.createPartitions(partitions, ignoreIfExists)
+ MsckFaultInjection.createCalls += 1
+ }
+
+ override def dropPartitions(partitions: JList[JMap[String, String]]): Unit = {
+ MsckFaultInjection.dropCalls += 1
+ if (MsckFaultInjection.failDrop) {
+ throw new IllegalStateException(MsckFaultInjection.DROP_FAILURE)
+ }
+ delegate.dropPartitions(partitions)
+ }
+
+ override def listPartitionsByNames(partitions: JList[JMap[String, String]]): JList[Partition] =
+ delegate.listPartitionsByNames(partitions)
+
+ override def listPartitions(prefix: JMap[String, String]): JList[Partition] = {
+ MsckFaultInjection.listCalls += 1
+ if (MsckFaultInjection.failList) {
+ throw new IllegalStateException(MsckFaultInjection.LIST_FAILURE)
+ }
+ delegate.listPartitions(prefix)
+ }
+}