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 a657beb510 [core][spark] Let DROP PARTITION repair overlapping Format
Table locations (#9712)
a657beb510 is described below
commit a657beb51014c5dbc51628459549d8f634e9e415
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Thu Sep 10 13:07:39 2026 +0800
[core][spark] Let DROP PARTITION repair overlapping Format Table locations
(#9712)
---
.../FormatTablePartitionRegistryValidator.java | 46 ++++++-
.../FormatTablePartitionRegistryValidatorTest.java | 148 +++++++++++++++++++++
.../paimon/spark/format/PaimonFormatTable.scala | 15 ++-
.../FormatTablePartitionManagementTest.scala | 70 ++++++++++
4 files changed, 273 insertions(+), 6 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
index 44c4f787af..1758389641 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidator.java
@@ -33,6 +33,7 @@ public final class FormatTablePartitionRegistryValidator {
private FormatTablePartitionRegistryValidator() {}
+ /** Validates each partition and rejects a registry whose partitions claim
each other's data. */
public static void validatePartitionLocations(
List<Partition> partitions,
List<String> partitionKeys,
@@ -40,6 +41,47 @@ public final class FormatTablePartitionRegistryValidator {
String tableName,
boolean onlyValueInPath,
@Nullable CatalogContext catalogContext) {
+ validate(
+ partitions,
+ partitionKeys,
+ tablePath,
+ tableName,
+ onlyValueInPath,
+ catalogContext,
+ true);
+ }
+
+ /**
+ * Validates every partition on its own: a complete spec, and a location
that parses and stays
+ * out of the table directory. Partitions are not compared to each other,
which is what an
+ * operation needs when it only has to know where each partition lives
rather than whether the
+ * registry as a whole is consistent.
+ */
+ public static void validateEachPartitionLocation(
+ List<Partition> partitions,
+ List<String> partitionKeys,
+ Path tablePath,
+ String tableName,
+ boolean onlyValueInPath,
+ @Nullable CatalogContext catalogContext) {
+ validate(
+ partitions,
+ partitionKeys,
+ tablePath,
+ tableName,
+ onlyValueInPath,
+ catalogContext,
+ false);
+ }
+
+ private static void validate(
+ List<Partition> partitions,
+ List<String> partitionKeys,
+ Path tablePath,
+ String tableName,
+ boolean onlyValueInPath,
+ @Nullable CatalogContext catalogContext,
+ boolean rejectPartitionsClaimingEachOther) {
FormatTablePartitionPathResolver resolver =
new FormatTablePartitionPathResolver(
tablePath, tableName, onlyValueInPath, catalogContext);
@@ -61,7 +103,9 @@ public final class FormatTablePartitionRegistryValidator {
resolver.resolve(
orderedSpec,
FormatTablePartitionPathResolver.customLocation(partition));
- resolver.validateAndRecord(orderedSpec, resolved);
+ if (rejectPartitionsClaimingEachOther) {
+ resolver.validateAndRecord(orderedSpec, resolved);
+ }
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidatorTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidatorTest.java
new file mode 100644
index 0000000000..0286ba48bf
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/format/FormatTablePartitionRegistryValidatorTest.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table.format;
+
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.partition.Partition;
+import org.apache.paimon.partition.PartitionStatistics;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.paimon.CoreOptions.PATH;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link FormatTablePartitionRegistryValidator}. */
+class FormatTablePartitionRegistryValidatorTest {
+
+ private static final List<String> PARTITION_KEYS = Arrays.asList("year",
"month");
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void
testPartitionsClaimingEachOtherAreRejectedOnlyWhenComparedToEachOther() {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "table");
+ Path external = new Path(new Path(tempDir.toUri()), "external");
+ List<Partition> registry =
+ Arrays.asList(
+ partitionAt(partitionSpec("2026", "01"),
external.toString()),
+ partitionAt(
+ partitionSpec("2026", "02"),
+ new Path(external, "child").toString()));
+
+ assertThatThrownBy(() -> validateTogether(registry, tablePath))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("overlapping locations");
+
+ // Each location is well formed on its own, so an operation that only
needs to know where
+ // every partition lives can go ahead - dropping one of the two is
what repairs the pair.
+ assertThatCode(() -> validateEach(registry,
tablePath)).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testLocationInsideTheTableDirectoryIsRejectedEitherWay() {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "table");
+ List<Partition> registry =
+ Collections.singletonList(
+ partitionAt(
+ partitionSpec("2026", "01"),
+ new Path(tablePath,
"year=2026/month=02").toString()));
+
+ assertThatThrownBy(() -> validateTogether(registry, tablePath))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("invalid custom location");
+ assertThatThrownBy(() -> validateEach(registry, tablePath))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("invalid custom location");
+ }
+
+ @Test
+ void testIncompleteSpecIsRejectedEitherWay() {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "table");
+ List<Partition> registry =
+ Collections.singletonList(
+ partitionAt(Collections.singletonMap("year", "2026"),
null));
+
+ assertThatThrownBy(() -> validateTogether(registry, tablePath))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("incomplete partition spec");
+ assertThatThrownBy(() -> validateEach(registry, tablePath))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("incomplete partition spec");
+ }
+
+ @Test
+ void testTwoRowsForOneSpecAreRejectedOnlyWhenComparedToEachOther() {
+ Path tablePath = new Path(new Path(tempDir.toUri()), "table");
+ Path external = new Path(new Path(tempDir.toUri()), "external");
+ List<Partition> registry =
+ Arrays.asList(
+ partitionAt(partitionSpec("2026", "01"),
external.toString()),
+ partitionAt(
+ partitionSpec("2026", "01"),
+ new Path(new Path(tempDir.toUri()),
"other").toString()));
+
+ assertThatThrownBy(() -> validateTogether(registry, tablePath))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("overlapping locations");
+ assertThatCode(() -> validateEach(registry,
tablePath)).doesNotThrowAnyException();
+ }
+
+ private static void validateTogether(List<Partition> registry, Path
tablePath) {
+ FormatTablePartitionRegistryValidator.validatePartitionLocations(
+ registry, PARTITION_KEYS, tablePath, "db.table", false, null);
+ }
+
+ private static void validateEach(List<Partition> registry, Path tablePath)
{
+ FormatTablePartitionRegistryValidator.validateEachPartitionLocation(
+ registry, PARTITION_KEYS, tablePath, "db.table", false, null);
+ }
+
+ private static Partition partitionAt(Map<String, String> spec, String
location) {
+ Map<String, String> options =
+ location == null ? null : Collections.singletonMap(PATH.key(),
location);
+ return new Partition(
+ spec,
+ 0,
+ 0,
+ 0,
+ 0,
+ PartitionStatistics.UNKNOWN_TOTAL_BUCKETS,
+ false,
+ null,
+ null,
+ null,
+ null,
+ options);
+ }
+
+ private static Map<String, String> partitionSpec(String year, String
month) {
+ LinkedHashMap<String, String> spec = new LinkedHashMap<>();
+ spec.put("year", year);
+ spec.put("month", month);
+ return spec;
+ }
+}
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 046a12015b..c610b9d031 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
@@ -362,10 +362,10 @@ case class PaimonFormatTable(table: FormatTable)
}
/**
- * Resolves DROP requests from one validated view of the catalog registry.
The boolean array is
- * aligned with the requests and tells callers which complete specs are
registered; partial-spec
- * entries are not used for existence reporting. The returned partitions are
the deduplicated
- * registered leaves covered by all requests.
+ * Resolves DROP requests from one view of the catalog registry, with every
partition validated on
+ * its own. The boolean array is aligned with the requests and tells callers
which complete specs
+ * are registered; partial-spec entries are not used for existence
reporting. The returned
+ * partitions are the deduplicated registered leaves covered by all requests.
*/
private[spark] def resolveFormatTablePartitionsForDrop(
partitionNames: Array[Array[String]],
@@ -385,7 +385,12 @@ case class PaimonFormatTable(table: FormatTable)
requested.foreach(spec =>
resolvePartitionPathWithinTable(orderedSpec(spec), onlyValueInPath))
val manager = requirePartitionManager()
val registry = manager.listPartitions(Collections.emptyMap[String,
String](), null)
- FormatTablePartitionRegistryValidator.validatePartitionLocations(
+ // A drop unregisters partitions and deletes default-location directories;
a custom location is
+ // never probed or deleted. Every partition still has to say where it
lives - a location inside
+ // the table directory could name a directory this drop deletes - but two
partitions that claim
+ // each other's data elsewhere do not change what this drop removes, and
refusing them here
+ // would leave that pair with no way out: dropping one of the two is how
it is repaired.
+ FormatTablePartitionRegistryValidator.validateEachPartitionLocation(
registry,
table.partitionKeys(),
new Path(table.location()),
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
index 06a152d2d6..5329e578ba 100644
---
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
@@ -577,6 +577,76 @@ class FormatTablePartitionManagementTest extends
SparkFunSuite {
}
}
+ test("catalog-managed DROP removes one of two partitions whose custom
locations overlap") {
+ val fileIO = LocalFileIO.create()
+ val tablePath =
+ new
Path(Files.createTempDirectory("catalog-partition-format-drop-overlapping-pair").toUri)
+ val outerDir =
+ new
Path(Files.createTempDirectory("catalog-partition-format-drop-overlapping-outer").toUri)
+ val nestedDir = new Path(outerDir, "child")
+ val outerFile = new Path(outerDir, "outer.csv")
+ val nestedFile = new Path(nestedDir, "nested.csv")
+ def customPartition(spec: Map[String, String], location: Path): Partition =
+ new Partition(
+ spec.asJava,
+ 0L,
+ 0L,
+ 0L,
+ 0L,
+ 0,
+ false,
+ null,
+ null,
+ null,
+ null,
+ Map(CoreOptions.PATH.key() -> location.toString).asJava)
+ val outerPartition = customPartition(partitionSpec(20260715, 10), outerDir)
+ val nestedPartition = customPartition(partitionSpec(20260716, 11),
nestedDir)
+ var dropped = Seq.empty[Map[String, String]]
+ val gateway = new FormatTablePartitionManager {
+ override def createPartitions(
+ partitions: JList[JMap[String, String]],
+ ignoreIfExists: Boolean,
+ statistics: JList[PartitionStatistics],
+ replaceStatistics: Boolean,
+ partitionOptions: JList[JMap[String, String]]): 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.singletonList(nestedPartition)
+
+ override def listPartitions(
+ prefix: JMap[String, String],
+ filter: Predicate): JList[Partition] =
+ Seq(outerPartition, nestedPartition).asJava
+ }
+
+ try {
+ fileIO.mkdirs(nestedDir)
+ fileIO.writeFile(outerFile, "outer", false)
+ fileIO.writeFile(nestedFile, "nested", false)
+ val sparkTable =
+ new PaimonFormatTable(
+ formatTableWithCatalogManagedPartitions(fileIO, tablePath.toString,
gateway))
+
+ // A registry where one custom location sits inside another cannot be
read, and dropping one
+ // of the two is the way out of it.
+ assert(
+ sparkTable
+ .dropFormatTablePartitions(Array(Array("dt", "hh")),
Array(partitionRow(20260716, 11))))
+
+ assert(dropped == Seq(Map("dt" -> "20260716", "hh" -> "11")))
+ assert(fileIO.exists(outerFile))
+ assert(fileIO.exists(nestedFile))
+ } finally {
+ fileIO.delete(tablePath, true)
+ fileIO.delete(outerDir, true)
+ }
+ }
+
test("catalog-managed DROP of a custom-location partition leaves all data in
place") {
val fileIO = LocalFileIO.create()
val tablePath =
