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 1c3df8f8db [spark] Reject empty partition values for format tables
(#9436)
1c3df8f8db is described below
commit 1c3df8f8db760d7acc68346f137aac63559db31a
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Fri Aug 28 13:48:54 2026 +0800
[spark] Reject empty partition values for format tables (#9436)
---
docs/docs/spark/sql-ddl.md | 6 ++
.../paimon/spark/format/PaimonFormatTable.scala | 40 +++++++-
.../CatalogManagedPartitionEdgeParityTest.scala | 114 ++++++++++++++++++++-
3 files changed, 156 insertions(+), 4 deletions(-)
diff --git a/docs/docs/spark/sql-ddl.md b/docs/docs/spark/sql-ddl.md
index e4ef5c707d..657ef33220 100644
--- a/docs/docs/spark/sql-ddl.md
+++ b/docs/docs/spark/sql-ddl.md
@@ -226,6 +226,12 @@ On a Format Table whose partitions are discovered from the
filesystem, `ADD PART
added partition before any data is written returns no rows. `DROP PARTITION`
unregisters the
partition and deletes its directory.
+A partition value that is empty or all whitespace is rejected by `ADD
PARTITION`, `DROP PARTITION`
+and `TRUNCATE PARTITION`. Such a value is written to the partition named by
+`partition.default-name` (`__DEFAULT_PARTITION__` unless configured
otherwise), the same partition
+a `NULL` is written to, so it names that partition rather than one of its own
- name it directly
+instead. Writing one through `INSERT` is unaffected and still lands there.
+
`ANALYZE TABLE` measures partitions. A Format Table has no snapshot to carry a
table-level
statistic and no column statistics, so `COMPUTE STATISTICS FOR COLUMNS` and
`FOR ALL COLUMNS` are
not supported on it; what the statement writes back to the catalog is the file
count, byte size,
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 baa7d21e40..08de978567 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
@@ -27,10 +27,11 @@ import org.apache.paimon.table.FormatTable
import org.apache.paimon.table.format.FormatTablePartitionManager
import org.apache.paimon.table.sink.BatchTableCommit
import org.apache.paimon.types.RowType
-import org.apache.paimon.utils.PartitionPathUtils
+import org.apache.paimon.utils.{PartitionPathUtils, StringUtils}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{NoSuchPartitionException,
NoSuchPartitionsException}
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.connector.catalog.{SupportsRead, SupportsWrite,
TableCapability, TableCatalog, TruncatableTable}
import org.apache.spark.sql.connector.catalog.TableCapability.{BATCH_READ,
BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC}
import org.apache.spark.sql.connector.distributions.Distribution
@@ -39,7 +40,7 @@ import org.apache.spark.sql.connector.read.ScanBuilder
import org.apache.spark.sql.connector.write._
import org.apache.spark.sql.connector.write.streaming.StreamingWrite
import org.apache.spark.sql.paimon.shims.SparkShimLoader
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{StringType, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import java.util
@@ -139,6 +140,8 @@ case class PaimonFormatTable(table: FormatTable)
return true
}
val partitionKeys = table.partitionKeys().asScala.toSeq
+ idents.foreach(
+ ident => requireNameablePartitionValues("TRUNCATE PARTITION", ident,
partitionKeys))
val specs = idents.map {
ident =>
require(
@@ -205,6 +208,35 @@ case class PaimonFormatTable(table: FormatTable)
requested.map(spec => registeredSpecs.contains(spec.asScala.toMap))
}
+ /**
+ * Rejects partition values the table cannot name. An empty or
whitespace-only string collapses to
+ * the default partition name on its way to the directory, the same name a
`NULL` gets, so the
+ * spec describes the null partition rather than one of its own. Adding it
registers a partition
+ * the value cannot round-trip to; dropping or truncating it hits the null
partition instead.
+ *
+ * `NULL` itself keeps its defined encoding and stays on that path.
+ */
+ private def requireNameablePartitionValues(
+ operation: String,
+ row: InternalRow,
+ partitionNames: Seq[String]): Unit = {
+ val fields = partitionSchema.fields.map(field => field.name -> field).toMap
+ partitionNames.take(row.numFields).zipWithIndex.foreach {
+ case (name, index) =>
+ val dataType =
CharVarcharUtils.replaceCharVarcharWithString(fields(name).dataType)
+ if (dataType == StringType && !row.isNullAt(index)) {
+ if (StringUtils.isNullOrWhitespaceOnly(row.getString(index))) {
+ val defaultPartitionName =
+ CoreOptions.fromMap(table.options()).partitionDefaultName()
+ throw new IllegalArgumentException(
+ s"$operation does not support an empty or whitespace-only string
for partition " +
+ s"column $name of Format Table ${table.fullName()}. Such a
value is written to " +
+ s"the partition named $defaultPartitionName, name it directly
to address it.")
+ }
+ }
+ }
+ }
+
private[spark] def createFormatTablePartitions(
rows: Array[InternalRow],
maps: Array[JMap[String, String]],
@@ -216,6 +248,7 @@ case class PaimonFormatTable(table: FormatTable)
val onlyValueInPath =
CoreOptions.fromMap(table.options()).formatTablePartitionOnlyValueInPath()
val partitionKeys = table.partitionKeys().asScala.toSeq
+ rows.foreach(row => requireNameablePartitionValues("ADD PARTITION", row,
partitionKeys))
val specs = rows.map(row => toPaimonPartition(row,
partitionKeys.take(row.numFields))).toSeq
// Resolve (and path-safety validate) every directory before mutating
anything.
val partitionPaths =
@@ -238,6 +271,9 @@ case class PaimonFormatTable(table: FormatTable)
partitionNames: Array[Array[String]],
rows: Array[InternalRow]): Boolean = {
val partitionKeyCount = table.partitionKeys().size()
+ rows.zip(partitionNames).foreach {
+ case (row, names) => requireNameablePartitionValues("DROP PARTITION",
row, names.toSeq)
+ }
val requested =
rows.zip(partitionNames).map { case (row, names) =>
toPaimonPartition(row, names.toSeq) }
val partitions = ArrayBuffer.empty[JMap[String, String]]
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 14333f57e5..e92bd629e3 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
@@ -211,6 +211,114 @@ class CatalogManagedPartitionEdgeParityTest extends
PaimonSparkTestWithRestCatal
}
}
+ private val unnameableValues = Seq("empty" -> "", "blank" -> " ")
+
+ Seq("STRING" -> "string", "CHAR(8)" -> "char", "VARCHAR(8)" ->
"varchar").foreach {
+ case (partitionType, typeSuffix) =>
+ unnameableValues.foreach {
+ case (valueLabel, value) =>
+ test(
+ s"ADD PARTITION rejects an $valueLabel $partitionType value
without changing state") {
+ val tableName = s"edge_reject_${valueLabel}_$typeSuffix"
+ withTable(tableName) {
+ createTable(tableName, partitionType)
+
+ val error = intercept[Exception] {
+ sql(s"ALTER TABLE ${qualified(tableName)} " +
+ s"ADD PARTITION (dt = '$value', hour = '00')").collect()
+ }
+
+ assert(
+ error.getMessage.contains(
+ "ADD PARTITION does not support an empty or whitespace-only
string"))
+ assert(error.getMessage.contains("partition column dt"))
+ assert(registered(tableName).isEmpty)
+ assert(!directoryExists(tableName,
s"dt=$defaultPartitionName/hour=00"))
+ }
+ }
+ }
+ }
+
+ test("ADD PARTITION names the offending column, wherever it sits in the
spec") {
+ val tableName = "edge_empty_second_column"
+ withTable(tableName) {
+ createTable(tableName)
+
+ val error = intercept[Exception] {
+ sql(
+ s"ALTER TABLE ${qualified(tableName)} " +
+ "ADD PARTITION (dt = '20260101', hour = '')").collect()
+ }
+
+ assert(error.getMessage.contains("partition column hour"))
+ assert(registered(tableName).isEmpty)
+ assert(!directoryExists(tableName,
s"dt=20260101/hour=$defaultPartitionName"))
+ }
+ }
+
+ test("one bad spec fails an ADD PARTITION batch before any of it is
applied") {
+ val tableName = "edge_empty_batch"
+ withTable(tableName) {
+ createTable(tableName)
+
+ val error = intercept[Exception] {
+ sql(
+ s"ALTER TABLE ${qualified(tableName)} ADD " +
+ "PARTITION (dt = '20260101', hour = '00') PARTITION (dt = '', hour
= '01')").collect()
+ }
+
+ assert(
+ error.getMessage.contains(
+ "ADD PARTITION does not support an empty or whitespace-only string"))
+ assert(registered(tableName).isEmpty)
+ assert(!directoryExists(tableName, "dt=20260101/hour=00"))
+ }
+ }
+
+ test("DROP PARTITION rejects an empty string instead of dropping the default
partition") {
+ val tableName = "edge_empty_drop"
+ withTable(tableName) {
+ createTable(tableName)
+ sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '', '00')")
+
+ val error = intercept[Exception] {
+ sql(
+ s"ALTER TABLE ${qualified(tableName)} " +
+ "DROP PARTITION (dt = '', hour = '00')").collect()
+ }
+
+ assert(
+ error.getMessage.contains(
+ "DROP PARTITION does not support an empty or whitespace-only
string"))
+ 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("TRUNCATE PARTITION rejects an empty string instead of emptying the
default partition") {
+ val tableName = "edge_empty_truncate"
+ withTable(tableName) {
+ createTable(tableName)
+ sql(s"INSERT INTO ${qualified(tableName)} VALUES (1, 'a', '', '00')")
+
+ val error = intercept[Exception] {
+ sql(
+ s"TRUNCATE TABLE ${qualified(tableName)} " +
+ "PARTITION (dt = '', hour = '00')").collect()
+ }
+
+ assert(
+ error.getMessage.contains(
+ "TRUNCATE PARTITION does not support an empty or whitespace-only
string"))
+ checkAnswer(
+ sql(s"SELECT id, payload, dt, hour FROM ${qualified(tableName)}"),
+ Seq(Row(1, "a", null, "00")))
+ }
+ }
+
test("an empty string partition value uses the default partition encoding") {
val tableName = "edge_empty_string"
withTable(tableName) {
@@ -267,8 +375,10 @@ class CatalogManagedPartitionEdgeParityTest extends
PaimonSparkTestWithRestCatal
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)
+ private def createTable(tableName: String): Unit = createTable(tableName,
"STRING")
+
+ private def createTable(tableName: String, partitionType: String): Unit =
+ sql(s"""CREATE TABLE $tableName (id INT, payload STRING, dt
$partitionType, hour STRING)
|USING CSV
|PARTITIONED BY (dt, hour)
|TBLPROPERTIES (
