From 745e96db720be01f655791618cbe71ee8cd34f37 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Sun, 9 Aug 2026 01:46:51 +0800 Subject: [PATCH 1/5] [spark] Cover the partition operations left over on format tables TRUNCATE TABLE, TRUNCATE TABLE PARTITION and ALTER TABLE ... RENAME TO PARTITION all refuse a format table with catalog-managed partitions, and these tests pin that they refuse without moving anything: a registration pointing at a directory that is gone, or data under a spec nobody registered, would both be worse than the refusal. The refusal TRUNCATE PARTITION produced described the table wrongly. "Only FileStoreTable supports partitions" is reached by a table that has partitions and lists them through this very trait, which sends the reader looking for the wrong problem; it now says which operations do manage those partitions. A null partition value registers under the default partition name, keeps a directory of that name, reads back as null, is discovered by MSCK REPAIR and is dropped by that name. SHOW PARTITIONS prints it as dt=null rather than the default name, which is what Spark's own ShowPartitionsExec does with a null value; a native Paimon table prints the same thing, so the test records where that behaviour comes from instead of working around it here. The remaining probes cover ADD/DROP PARTITION idempotence, prefix DROP, partial specs in SHOW PARTITIONS, case-insensitive partition column names and empty string partition values, all of which already behaved this way. --- .../spark/PaimonPartitionManagement.scala | 8 +- ...CatalogManagedPartitionDdlParityTest.scala | 184 +++++++++++ ...atalogManagedPartitionEdgeParityTest.scala | 309 ++++++++++++++++++ 3 files changed, 500 insertions(+), 1 deletion(-) create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionDdlParityTest.scala create mode 100644 paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala 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 3f847c1252d2..95acdc5cd14d 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 @@ -58,7 +58,13 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement with L toPaimonPartition(r, partitionKeys.take(r.numFields)) } case _ => - throw new UnsupportedOperationException("Only FileStoreTable supports partitions.") + // Reached by the partition operations this trait still serves directly, such as TRUNCATE + // PARTITION. Saying that only a FileStoreTable has partitions would be wrong for a Format + // Table with catalog-managed partitions, which has them and lists them here. + throw new UnsupportedOperationException( + s"This partition operation is supported only for a Paimon table, and ${table.name()} is " + + s"not one. A Format Table manages its partitions through ADD PARTITION, " + + s"DROP PARTITION and MSCK REPAIR TABLE.") } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionDdlParityTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionDdlParityTest.scala new file mode 100644 index 000000000000..4ce81c53db14 --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionDdlParityTest.scala @@ -0,0 +1,184 @@ +/* + * 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.spark.PaimonSparkTestWithRestCatalogBase +import org.apache.paimon.table.FormatTable + +import org.apache.spark.sql.Row + +import scala.collection.JavaConverters._ + +/** + * Partition DDL on a Format Table with catalog-managed partitions, held against what Spark's own + * `AlterTableAddPartitionSuite`, `AlterTableDropPartitionSuite` and `ShowPartitionsSuite` pin for a + * metastore table: `IF NOT EXISTS` / `IF EXISTS` decide whether a repeat is an error, `SHOW + * PARTITIONS` takes a partial spec and sorts its output, and partition column names resolve under + * the session's case sensitivity. + */ +class CatalogManagedPartitionDdlParityTest extends PaimonSparkTestWithRestCatalogBase { + + test("ADD PARTITION IF NOT EXISTS is a repeatable no-op, a strict repeat is an error") { + val tableName = "ddl_add_if_not_exists" + withTable(tableName) { + createTable(tableName) + + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '00')") + sql( + s"ALTER TABLE ${qualified(tableName)} ADD IF NOT EXISTS " + + s"PARTITION (dt = '20260101', hour = '00')") + assert(registered(tableName) == Set("20260101/00")) + + // Spark reports a duplicate as PartitionsAlreadyExistException; the point being pinned is + // that a strict ADD does not quietly succeed. + val error = intercept[Exception] { + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '00')") + } + assert(causeMessages(error).contains("20260101"), causeMessages(error)) + assert(registered(tableName) == Set("20260101/00")) + } + } + + test("DROP PARTITION IF EXISTS tolerates a partition that is not there") { + val tableName = "ddl_drop_if_exists" + withTable(tableName) { + createTable(tableName) + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '00')") + + sql( + s"ALTER TABLE ${qualified(tableName)} DROP IF EXISTS " + + s"PARTITION (dt = '20991231', hour = '00')") + assert(registered(tableName) == Set("20260101/00")) + + sql(s"ALTER TABLE ${qualified(tableName)} DROP PARTITION (dt = '20260101', hour = '00')") + assert(registered(tableName).isEmpty) + } + } + + test("ADD and DROP resolve partition column names the way the rest of Spark resolves them") { + val tableName = "ddl_case_insensitive" + withTable(tableName) { + createTable(tableName) + + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (DT = '20260101', HOUR = '00')") + assert(registered(tableName) == Set("20260101/00")) + + sql(s"ALTER TABLE ${qualified(tableName)} DROP PARTITION (Dt = '20260101', Hour = '00')") + assert(registered(tableName).isEmpty) + } + } + + test("DROP with a leading prefix removes every partition under it and leaves the rest") { + val tableName = "ddl_drop_prefix" + withTable(tableName) { + createTable(tableName) + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '00')") + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '01')") + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260102', hour = '00')") + + sql(s"ALTER TABLE ${qualified(tableName)} DROP PARTITION (dt = '20260101')") + + assert(registered(tableName) == Set("20260102/00")) + } + } + + test("SHOW PARTITIONS takes a partial spec and returns sorted output") { + val tableName = "ddl_show_partitions" + withTable(tableName) { + createTable(tableName) + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260102', hour = '01')") + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '01')") + sql(s"ALTER TABLE ${qualified(tableName)} ADD PARTITION (dt = '20260101', hour = '00')") + + val all = sql(s"SHOW PARTITIONS ${qualified(tableName)}").collect().map(_.getString(0)).toSeq + assert(all == all.sorted, all.mkString(", ")) + assert( + all == Seq("dt=20260101/hour=00", "dt=20260101/hour=01", "dt=20260102/hour=01"), + all.mkString(", ")) + + val scoped = sql(s"SHOW PARTITIONS ${qualified(tableName)} PARTITION (dt = '20260101')") + .collect() + .map(_.getString(0)) + .toSet + assert(scoped == Set("dt=20260101/hour=00", "dt=20260101/hour=01"), scoped.mkString(", ")) + } + } + + test("an INSERT registers the partition it wrote and the row reads back") { + val tableName = "ddl_insert_registers" + withTable(tableName) { + createTable(tableName) + + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + + assert(registered(tableName) == Set("20260101/00")) + checkAnswer( + sql(s"SELECT id, payload, dt, hour FROM ${qualified(tableName)}"), + Seq(Row(1, "a", "20260101", "00"))) + } + } + + test("a registered partition whose directory is gone reads as empty rather than failing") { + val tableName = "ddl_missing_directory" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + sql(s"INSERT INTO ${qualified(tableName)} VALUES (2, 'b', '20260102', '00')") + + val table = formatTable(tableName) + table + .fileIO() + .deleteDirectoryQuietly(new org.apache.paimon.fs.Path(table.location(), "dt=20260101")) + + // The registration is the authority on which partitions exist; a missing directory is drift, + // and drift reads as empty instead of taking the query down. + checkAnswer(sql(s"SELECT id FROM ${qualified(tableName)} ORDER BY id"), Seq(Row(2))) + assert(registered(tableName) == Set("20260101/00", "20260102/00")) + } + } + + private def qualified(tableName: String): String = s"paimon.$dbName0.$tableName" + + private def createTable(tableName: String): Unit = + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING CSV + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + + private def formatTable(tableName: String): FormatTable = + paimonCatalog.getTable(Identifier.create(dbName0, tableName)).asInstanceOf[FormatTable] + + private def registered(tableName: String): Set[String] = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .map(p => s"${p.spec().get("dt")}/${p.spec().get("hour")}") + .toSet + + private def causeMessages(error: Throwable): String = + Iterator + .iterate(error)(_.getCause) + .takeWhile(_ != null) + .map(e => String.valueOf(e.getMessage)) + .mkString(" | ") +} diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala new file mode 100644 index 000000000000..9a16f5751aad --- /dev/null +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala @@ -0,0 +1,309 @@ +/* + * 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.spark.PaimonSparkTestWithRestCatalogBase +import org.apache.paimon.table.FormatTable + +import org.apache.spark.sql.Row + +import scala.collection.JavaConverters._ + +/** + * The partition operations left over once ADD, DROP, SHOW and ANALYZE are covered: `TRUNCATE`, + * `RENAME PARTITION`, and a null or empty string as a partition value. Spark pins all three for a + * metastore table in `TruncateTableSuiteBase`, `AlterTableRenamePartitionSuiteBase` and the + * `SPARK-33591` / `SPARK-33904` cases of the add, drop and show partition suites. + * + * An operation this table cannot support has to be refused outright. The failure worth catching is + * a half-done one: the registration moved and the directory did not, or the data went away and the + * registration stayed behind claiming it is still there. + */ +class CatalogManagedPartitionEdgeParityTest extends PaimonSparkTestWithRestCatalogBase { + + private val defaultPartitionName = "__DEFAULT_PARTITION__" + + // ------------------------------------------------------------------ TRUNCATE + + test("TRUNCATE TABLE either clears the data and keeps the partitions, or refuses outright") { + val tableName = "edge_truncate_table" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + sql(s"INSERT INTO ${qualified(tableName)} VALUES (2, 'b', '20260102', '00')") + val before = registered(tableName) + + val outcome = attempt(s"TRUNCATE TABLE ${qualified(tableName)}") + + outcome match { + case Refused(_) => + // Refusing is a defensible answer; leaving the table half-truncated is not. + assert(registered(tableName) == before) + assert(rowIds(tableName) == Seq(1, 2)) + case Accepted => + // Spark keeps the partitions of a truncated table (SPARK-34418): truncation empties a + // table, it does not redefine which partitions it has. + assert(registered(tableName) == before, "TRUNCATE dropped partition registrations") + assert(rowIds(tableName).isEmpty, "TRUNCATE left rows behind") + } + } + } + + test("TRUNCATE TABLE PARTITION stays inside the partition it names, or refuses outright") { + val tableName = "edge_truncate_partition" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + sql(s"INSERT INTO ${qualified(tableName)} VALUES (2, 'b', '20260102', '00')") + val before = registered(tableName) + + val outcome = + attempt(s"TRUNCATE TABLE ${qualified(tableName)} PARTITION (dt = '20260101', hour = '00')") + + outcome match { + case Refused(message) => + assert(registered(tableName) == before) + assert(rowIds(tableName) == Seq(1, 2)) + // The refusal has to name what is unsupported. This table has partitions and lists them, + // so a message claiming otherwise sends the reader looking for the wrong problem. + assert( + !message.contains("Only FileStoreTable supports partitions"), + s"refusal misdescribes the table: $message") + assert(message.contains("MSCK REPAIR TABLE"), message) + case Accepted => + assert(registered(tableName) == before, "TRUNCATE PARTITION dropped registrations") + assert(rowIds(tableName) == Seq(2), "TRUNCATE PARTITION touched the wrong partitions") + } + } + } + + // ------------------------------------------------------------------ RENAME PARTITION + + test("RENAME PARTITION moves both the registration and the data, or refuses outright") { + val tableName = "edge_rename_partition" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '20260101', '00')") + val before = registered(tableName) + + val outcome = attempt( + s"ALTER TABLE ${qualified(tableName)} PARTITION (dt = '20260101', hour = '00') " + + s"RENAME TO PARTITION (dt = '20260201', hour = '00')") + + outcome match { + case Refused(_) => + // Nothing may have moved: a registration pointing at a directory that is no longer + // there, or data under a spec nobody registered, are both worse than a refusal. + assert(registered(tableName) == before) + assert(rowIds(tableName) == Seq(1)) + assert(directoryExists(tableName, "dt=20260101/hour=00")) + case Accepted => + assert(registered(tableName) == Set("20260201/00")) + assert(rowIds(tableName) == Seq(1), "renamed partition lost its rows") + assert(directoryExists(tableName, "dt=20260201/hour=00")) + assert(!directoryExists(tableName, "dt=20260101/hour=00")) + } + } + } + + // ------------------------------------------------------------------ null / empty partition values + + test("a null partition value registers under the default partition name and reads back as null") { + val tableName = "edge_null_partition" + withTable(tableName) { + createTable(tableName) + + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', NULL, '00')") + + assert(registered(tableName) == Set(s"$defaultPartitionName/00")) + assert(directoryExists(tableName, s"dt=$defaultPartitionName/hour=00")) + checkAnswer( + sql(s"SELECT id, payload, dt, hour FROM ${qualified(tableName)}"), + Seq(Row(1, "a", null, "00"))) + } + } + + test("SHOW PARTITIONS spells a null partition value the way Spark's v2 path spells it") { + val tableName = "edge_null_show" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', NULL, '00')") + sql(s"INSERT INTO ${qualified(tableName)} VALUES (2, 'b', '20260101', '00')") + + // Not what Hive prints. Spark's own ShowPartitionsExec renders a null partition value as the + // literal "null" (`if (partValueUTF8String == null) "null"`), while its v1 path over a Hive + // table prints __HIVE_DEFAULT_PARTITION__ because the metastore spec already holds that + // string. Both the registration and the directory here use the default partition name, so + // what SHOW prints is not a spec that can be pasted back into DROP PARTITION — see the + // drop-by-name case below for the spelling that does work. + val shown = + sql(s"SHOW PARTITIONS ${qualified(tableName)}").collect().map(_.getString(0)).toSet + assert(shown == Set("dt=null/hour=00", "dt=20260101/hour=00"), shown.mkString(", ")) + assert(registered(tableName).contains(s"$defaultPartitionName/00")) + } + } + + test("the null rendering is Spark's, not this table's: a native Paimon table prints it too") { + val tableName = "edge_null_show_paimon" + withTable(tableName) { + sql(s"""CREATE TABLE $tableName (id INT, dt STRING) + |PARTITIONED BY (dt) + |""".stripMargin) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, NULL)") + + val shown = + sql(s"SHOW PARTITIONS ${qualified(tableName)}").collect().map(_.getString(0)).toSet + // Same rendering on a table that has nothing to do with Format Tables, which puts the + // divergence in Spark's v2 command rather than in the catalog-managed partition work. + assert(shown == Set("dt=null"), shown.mkString(", ")) + } + } + + test("MSCK discovers a default partition directory an outside writer left") { + val tableName = "edge_null_msck" + withTable(tableName) { + createTable(tableName) + writeCsvPartition(tableName, defaultPartitionName, "00", 7) + + sql(s"MSCK REPAIR TABLE ${qualified(tableName)}").collect() + + assert(registered(tableName) == Set(s"$defaultPartitionName/00")) + // The registration is only worth something if the rows behind it are readable. + assert(rowIds(tableName) == Seq(7)) + } + } + + test("the default partition can be dropped by name and takes its data with it") { + val tableName = "edge_null_drop" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', NULL, '00')") + sql(s"INSERT INTO ${qualified(tableName)} VALUES (2, 'b', '20260101', '00')") + + sql( + s"ALTER TABLE ${qualified(tableName)} DROP PARTITION " + + s"(dt = '$defaultPartitionName', hour = '00')") + + assert(registered(tableName) == Set("20260101/00")) + assert(!directoryExists(tableName, s"dt=$defaultPartitionName/hour=00")) + assert(rowIds(tableName) == Seq(2)) + } + } + + test("ANALYZE measures the default partition like any other") { + val tableName = "edge_null_analyze" + withTable(tableName) { + createTable(tableName) + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', NULL, '00')") + + sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() + + val measured = paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .head + assert(measured.fileCount() == 1L, measured.toString) + assert(measured.fileSizeInBytes() > 0L, measured.toString) + } + } + + test("an empty string partition value is not confused with a null one") { + val tableName = "edge_empty_string" + withTable(tableName) { + createTable(tableName) + + sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '', '00')") + + // Whichever way an empty value is spelled on disk, the row has to read back the way it was + // written, and the partition it landed in has to be the one that is registered. + val registrations = registered(tableName) + assert(registrations.size == 1, registrations.mkString(", ")) + val Array(dt, _) = registrations.head.split("/", 2) + assert(directoryExists(tableName, s"dt=$dt/hour=00"), s"no directory for registered dt=$dt") + assert(rowIds(tableName) == Seq(1)) + } + } + + // ------------------------------------------------------------------ helpers + + sealed private trait Outcome + private case object Accepted extends Outcome + private case class Refused(message: String) extends Outcome + + /** Runs a statement, reporting whether it was accepted rather than failing the test outright. */ + private def attempt(statement: String): Outcome = + try { + sql(statement).collect() + Accepted + } catch { + case error: Throwable => + val message = causeMessages(error) + // scalastyle:off println + println(s"[edge-parity] refused: $statement -> $message") + // scalastyle:on println + Refused(message) + } + + private def qualified(tableName: String): String = s"paimon.$dbName0.$tableName" + + private def createTable(tableName: String): Unit = + sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt STRING, hour STRING) + |USING CSV + |PARTITIONED BY (dt, hour) + |TBLPROPERTIES ( + | 'format-table.implementation' = 'paimon', + | 'metastore.partitioned-table' = 'true') + |""".stripMargin) + + private def formatTable(tableName: String): FormatTable = + paimonCatalog.getTable(Identifier.create(dbName0, tableName)).asInstanceOf[FormatTable] + + private def directoryExists(tableName: String, relativePath: String): Boolean = { + val table = formatTable(tableName) + table.fileIO().exists(new Path(table.location(), relativePath)) + } + + private def writeCsvPartition(tableName: String, dt: String, hour: String, id: Int): Unit = { + val table = formatTable(tableName) + val partitionPath = new Path(table.location(), s"dt=$dt/hour=$hour") + table.fileIO().mkdirs(partitionPath) + table + .fileIO() + .writeFile(new Path(partitionPath, f"part-$id%05d.csv"), s"$id,payload-$id\n", false) + } + + private def registered(tableName: String): Set[String] = + paimonCatalog + .listPartitions(Identifier.create(dbName0, tableName)) + .asScala + .map(p => s"${p.spec().get("dt")}/${p.spec().get("hour")}") + .toSet + + private def rowIds(tableName: String): Seq[Int] = + sql(s"SELECT id FROM ${qualified(tableName)} ORDER BY id").collect().map(_.getInt(0)).toSeq + + private def causeMessages(error: Throwable): String = + Iterator + .iterate(error)(_.getCause) + .takeWhile(_ != null) + .map(e => String.valueOf(e.getMessage)) + .mkString(" | ") +} From 2d213ef3b4b15d4de1b3ab0f628b735b4ca02b4d Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 11 Aug 2026 14:28:07 +0800 Subject: [PATCH 2/5] [spark] Keep edge parity tests independent from ANALYZE --- .../CatalogManagedPartitionEdgeParityTest.scala | 17 ----------------- 1 file changed, 17 deletions(-) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala index 9a16f5751aad..021540b9ea8f 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala @@ -208,23 +208,6 @@ class CatalogManagedPartitionEdgeParityTest extends PaimonSparkTestWithRestCatal } } - test("ANALYZE measures the default partition like any other") { - val tableName = "edge_null_analyze" - withTable(tableName) { - createTable(tableName) - sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', NULL, '00')") - - sql(s"ANALYZE TABLE ${qualified(tableName)} COMPUTE STATISTICS NOSCAN").collect() - - val measured = paimonCatalog - .listPartitions(Identifier.create(dbName0, tableName)) - .asScala - .head - assert(measured.fileCount() == 1L, measured.toString) - assert(measured.fileSizeInBytes() > 0L, measured.toString) - } - } - test("an empty string partition value is not confused with a null one") { val tableName = "edge_empty_string" withTable(tableName) { From 16bb2188c5dca2408b6ed156639e0ec69690e2ee Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 11 Aug 2026 15:07:07 +0800 Subject: [PATCH 3/5] [spark] Assert empty partition encoding in edge tests --- .../CatalogManagedPartitionEdgeParityTest.scala | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala index 021540b9ea8f..e40f732d2855 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala @@ -208,20 +208,20 @@ class CatalogManagedPartitionEdgeParityTest extends PaimonSparkTestWithRestCatal } } - test("an empty string partition value is not confused with a null one") { + test("an empty string partition value uses the default partition encoding") { val tableName = "edge_empty_string" withTable(tableName) { createTable(tableName) sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '', '00')") - // Whichever way an empty value is spelled on disk, the row has to read back the way it was - // written, and the partition it landed in has to be the one that is registered. - val registrations = registered(tableName) - assert(registrations.size == 1, registrations.mkString(", ")) - val Array(dt, _) = registrations.head.split("/", 2) - assert(directoryExists(tableName, s"dt=$dt/hour=00"), s"no directory for registered dt=$dt") - assert(rowIds(tableName) == Seq(1)) + // Paimon's partition encoding normalizes a null or empty dynamic partition value to the + // configured default partition name, and the reader decodes it as null. + assert(registered(tableName) == Set(s"$defaultPartitionName/00")) + assert(directoryExists(tableName, s"dt=$defaultPartitionName/hour=00")) + checkAnswer( + sql(s"SELECT id, payload, dt, hour FROM ${qualified(tableName)}"), + Seq(Row(1, "a", null, "00"))) } } From 6ea313b774e716117fb35350ac60c1fadcf7b937 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Tue, 11 Aug 2026 21:18:53 +0800 Subject: [PATCH 4/5] [spark] Address review: native Paimon table wording and strict refusal check --- .../spark/PaimonPartitionManagement.scala | 17 ++++++++----- ...atalogManagedPartitionEdgeParityTest.scala | 24 +++++++++++++++++-- 2 files changed, 33 insertions(+), 8 deletions(-) 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 95acdc5cd14d..7bf95789899c 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 @@ -20,7 +20,7 @@ package org.apache.paimon.spark import org.apache.paimon.CoreOptions import org.apache.paimon.partition.PartitionStatistics -import org.apache.paimon.table.{FileStoreTable, Table} +import org.apache.paimon.table.{FileStoreTable, FormatTable, Table} import org.apache.paimon.table.source.ScanMode import org.apache.paimon.types.RowType import org.apache.paimon.utils.{InternalRowPartitionComputer, TypeUtils} @@ -57,14 +57,19 @@ trait PaimonPartitionManagement extends SupportsAtomicPartitionManagement with L ) toPaimonPartition(r, partitionKeys.take(r.numFields)) } - case _ => + case _: FormatTable => // Reached by the partition operations this trait still serves directly, such as TRUNCATE // PARTITION. Saying that only a FileStoreTable has partitions would be wrong for a Format - // Table with catalog-managed partitions, which has them and lists them here. + // Table with catalog-managed partitions, which has them and lists them here; a Format + // Table is still a Paimon table, just not a native one. + throw new UnsupportedOperationException( + s"This partition operation is supported only for a native Paimon table; " + + s"${table.name()} is a Format Table, which manages its partitions through " + + s"ADD PARTITION, DROP PARTITION and MSCK REPAIR TABLE.") + case _ => throw new UnsupportedOperationException( - s"This partition operation is supported only for a Paimon table, and ${table.name()} is " + - s"not one. A Format Table manages its partitions through ADD PARTITION, " + - s"DROP PARTITION and MSCK REPAIR TABLE.") + s"This partition operation is supported only for a native Paimon table, " + + s"which ${table.name()} is not.") } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala index e40f732d2855..14333f57e58d 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/CatalogManagedPartitionEdgeParityTest.scala @@ -25,7 +25,10 @@ import org.apache.paimon.table.FormatTable import org.apache.spark.sql.Row +import java.util.Locale + import scala.collection.JavaConverters._ +import scala.util.control.NonFatal /** * The partition operations left over once ADD, DROP, SHOW and ANALYZE are covered: `TRUNCATE`, @@ -231,13 +234,18 @@ class CatalogManagedPartitionEdgeParityTest extends PaimonSparkTestWithRestCatal private case object Accepted extends Outcome private case class Refused(message: String) extends Outcome - /** Runs a statement, reporting whether it was accepted rather than failing the test outright. */ + /** + * Runs a statement, reporting whether it was accepted rather than failing the test outright. Only + * an intentional refusal counts as [[Refused]]; any other failure, such as a parse or resolution + * error, would also leave the state unchanged, so classifying it as a refusal would let a test + * pass without exercising the command it meant to. Such a failure propagates and fails the test. + */ private def attempt(statement: String): Outcome = try { sql(statement).collect() Accepted } catch { - case error: Throwable => + case NonFatal(error) if isExpectedRefusal(error) => val message = causeMessages(error) // scalastyle:off println println(s"[edge-parity] refused: $statement -> $message") @@ -245,6 +253,18 @@ class CatalogManagedPartitionEdgeParityTest extends PaimonSparkTestWithRestCatal Refused(message) } + /** + * An intentional refusal is Paimon's own [[UnsupportedOperationException]] somewhere in the cause + * chain, or Spark declining the command for a table that does not support it. + */ + private def isExpectedRefusal(error: Throwable): Boolean = { + val causes = Iterator.iterate(error)(_.getCause).takeWhile(_ != null).toSeq + causes.exists(_.isInstanceOf[UnsupportedOperationException]) || { + val message = causeMessages(error).toLowerCase(Locale.ROOT) + message.contains("not supported") || message.contains("does not support") + } + } + private def qualified(tableName: String): String = s"paimon.$dbName0.$tableName" private def createTable(tableName: String): Unit = From febbaddea0f0a3aeffd6f676967cce7d7c81bf02 Mon Sep 17 00:00:00 2001 From: Sun Dapeng Date: Wed, 12 Aug 2026 14:58:37 +0800 Subject: [PATCH 5/5] [spark] Add remediation hints to Format Table partition DDL errors --- .../execution/PaimonFormatTablePartitionDdlExec.scala | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) 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 index a49dab7019a0..45c925b4469d 100644 --- 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 @@ -18,6 +18,7 @@ package org.apache.paimon.spark.execution +import org.apache.paimon.CoreOptions import org.apache.paimon.spark.format.{FormatTablePartitionRepair, PaimonFormatTable} import org.apache.spark.sql.catalyst.InternalRow @@ -63,7 +64,11 @@ object PaimonFormatTablePartitionDdlExec { 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.") + s"partitions, but partitions of ${table.name()} are discovered from the filesystem. " + + s"A partition of this table is a plain directory under the table location: create or " + + s"remove the directory directly, or enable catalog-managed partitions with ALTER TABLE " + + s"... SET TBLPROPERTIES ('${CoreOptions.METASTORE_PARTITIONED_TABLE.key()}' = 'true') " + + s"followed by MSCK REPAIR TABLE to register the existing partitions.") } /**