This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new ff666513774b [SPARK-51830][4.2][SQL] Fix partition type mismatch when
validatePartitionColumns is disabled
ff666513774b is described below
commit ff666513774b89cc7d15f090e932c9867ec95728
Author: Szehon Ho <[email protected]>
AuthorDate: Wed Sep 30 20:14:12 2026 -0700
[SPARK-51830][4.2][SQL] Fix partition type mismatch when
validatePartitionColumns is disabled
### What changes were proposed in this pull request?
Backport #57994 (commit 7e8bf895d1cf21676b021a5b225e2e3c1fc3ea70) to
`branch-4.2` for SPARK-51830. The commit cherry-picks cleanly, with the same
production changes and regression tests.
The write command passes `spark.sql.sources.validatePartitionColumns` when
constructing partition paths. When validation is disabled, nonnumeric legacy
partition values are preserved instead of throwing `NumberFormatException`.
### Why are the changes needed?
An existing nonnumeric partition value in a table with a numeric partition
schema can prevent valid inserts even when partition validation is explicitly
disabled. This restores the configured behavior for both dynamic inserts and
static partition metadata refreshes on Spark 4.2.
### Does this PR introduce _any_ user-facing change?
Yes. Affected inserts succeed with
`spark.sql.sources.validatePartitionColumns=false`. Default strict validation
and normalization of valid numeric partition values remain unchanged.
### How was this patch tested?
Ran the four SPARK-51830 regression tests locally on the backport: all
passed across `PartitioningUtilsSuite`, `ParquetV2PartitionDiscoverySuite`, and
`InsertSuite`.
```sh
SPARK_LOCAL_IP=127.0.0.1 build/sbt -Dsbt.log.noformat=true 'sql/testOnly
org.apache.spark.sql.execution.datasources.PartitioningUtilsSuite
org.apache.spark.sql.execution.datasources.parquet.ParquetV2PartitionDiscoverySuite
org.apache.spark.sql.sources.InsertSuite -- -z SPARK-51830'
```
The backport's added and removed lines match the original commit in all
five files. `git diff --check` and checks of added lines for ASCII and line
length pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Codex prepared this backport and PR description. The original patch retains
its author attribution.
Closes #59173 from szehon-ho/codex/backport-spark-51830-4.2.
Lead-authored-by: Szehon Ho <[email protected]>
Co-authored-by: Pratham Manja <[email protected]>
Signed-off-by: Szehon Ho <[email protected]>
---
.../InsertIntoHadoopFsRelationCommand.scala | 7 +-
.../execution/datasources/PartitioningUtils.scala | 41 +++++++-
.../datasources/PartitioningUtilsSuite.scala | 104 +++++++++++++++++++++
.../parquet/ParquetPartitionDiscoverySuite.scala | 11 +++
.../org/apache/spark/sql/sources/InsertSuite.scala | 56 ++++++++++-
5 files changed, 211 insertions(+), 8 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
index 03d4f09ab337..d969dbcd8ea9 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InsertIntoHadoopFsRelationCommand.scala
@@ -200,7 +200,8 @@ case class InsertIntoHadoopFsRelationCommand(
&& partitionColumns.length == staticPartitions.size) {
// Avoid empty static partition can't loaded to datasource table.
val staticPathFragment =
- PartitioningUtils.getPathFragment(staticPartitions, partitionColumns)
+ PartitioningUtils.getPathFragment(
+ staticPartitions, partitionColumns, conf.validatePartitionColumns)
refreshUpdatedPartitions(Set(staticPathFragment))
} else {
refreshUpdatedPartitions(updatedPartitionPaths)
@@ -267,9 +268,11 @@ case class InsertIntoHadoopFsRelationCommand(
table: CatalogTable,
qualifiedOutputPath: Path,
partitions: Seq[CatalogTablePartition]): Map[TablePartitionSpec, String]
= {
+ val validatePartitionColumns = conf.validatePartitionColumns
partitions.flatMap { p =>
val defaultLocation = qualifiedOutputPath.suffix(
- "/" + PartitioningUtils.getPathFragment(p.spec,
table.partitionSchema)).toString
+ "/" + PartitioningUtils.getPathFragment(
+ p.spec, table.partitionSchema, validatePartitionColumns)).toString
val catalogLocation = new Path(p.location).makeQualified(
fs.getUri, fs.getWorkingDirectory).toString
if (catalogLocation != defaultLocation) {
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
index 2583ff671fed..16ae22d611f7 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
@@ -357,23 +357,54 @@ object PartitioningUtils extends SQLConfHelper {
/**
* This is the inverse of parsePathFragment().
*/
- def getPathFragment(spec: TablePartitionSpec, partitionSchema: StructType):
String = {
+ def getPathFragment(
+ spec: TablePartitionSpec,
+ partitionSchema: StructType,
+ validatePartitionColumns: Boolean): String = {
partitionSchema.map { field =>
escapePathName(field.name) + "=" +
getPartitionValueString(
- removeLeadingZerosFromNumberTypePartition(spec(field.name),
field.dataType))
+ removeLeadingZerosFromNumberTypePartition(
+ spec(field.name), field.dataType, validatePartitionColumns))
}.mkString("/")
}
- def removeLeadingZerosFromNumberTypePartition(value: String, dataType:
DataType): String =
+ def getPathFragment(spec: TablePartitionSpec, partitionSchema: StructType):
String = {
+ getPathFragment(spec, partitionSchema, validatePartitionColumns = true)
+ }
+
+ def removeLeadingZerosFromNumberTypePartition(
+ value: String,
+ dataType: DataType,
+ validatePartitionColumns: Boolean): String =
dataType match {
case ByteType | ShortType | IntegerType | LongType | FloatType |
DoubleType =>
- Option(castPartValueToDesiredType(dataType, value,
null)).map(_.toString).orNull
+ if (value == null) {
+ null
+ } else {
+ try {
+ Option(castPartValueToDesiredType(dataType, value,
null)).map(_.toString).orNull
+ } catch {
+ case _: NumberFormatException if !validatePartitionColumns =>
+ // If validation is disabled and value cannot be cast to numeric
type, return as-is.
+ // This handles legacy tables with string partition values that
don't match
+ // the partition column's numeric type definition.
+ // Leading zeros (e.g., "007") are still normalized to "7" since
they parse.
+ value
+ }
+ }
case _ => value
}
+ def getPathFragment(
+ spec: TablePartitionSpec,
+ partitionColumns: Seq[Attribute],
+ validatePartitionColumns: Boolean): String = {
+ getPathFragment(spec, DataTypeUtils.fromAttributes(partitionColumns),
validatePartitionColumns)
+ }
+
def getPathFragment(spec: TablePartitionSpec, partitionColumns:
Seq[Attribute]): String = {
- getPathFragment(spec, DataTypeUtils.fromAttributes(partitionColumns))
+ getPathFragment(spec, partitionColumns, validatePartitionColumns = true)
}
/**
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PartitioningUtilsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PartitioningUtilsSuite.scala
new file mode 100644
index 000000000000..64c0a5fe217b
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PartitioningUtilsSuite.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.execution.datasources
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.types._
+
+class PartitioningUtilsSuite extends SparkFunSuite {
+
+ test("SPARK-51830: removeLeadingZerosFromNumberTypePartition with validation
disabled") {
+ // Test with validation enabled (default behavior)
+ // String values that cannot be cast to numeric types should throw
NumberFormatException
+ intercept[NumberFormatException] {
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "abc", IntegerType, validatePartitionColumns = true)
+ }
+
+ intercept[NumberFormatException] {
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "xyz", LongType, validatePartitionColumns = true)
+ }
+
+ intercept[NumberFormatException] {
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "invalid", DoubleType, validatePartitionColumns = true)
+ }
+
+ // Test with validation disabled
+ // String values should be preserved as-is for legacy compatibility
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "abc", IntegerType, validatePartitionColumns = false) === "abc"
+ )
+
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "xyz", LongType, validatePartitionColumns = false) === "xyz"
+ )
+
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "invalid", DoubleType, validatePartitionColumns = false) === "invalid"
+ )
+
+ // Valid numeric strings should still be normalized when validation is
disabled
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "42", IntegerType, validatePartitionColumns = false) === "42"
+ )
+
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "9223372036854775807", LongType, validatePartitionColumns = false) ===
"9223372036854775807"
+ )
+
+ // Leading zeros should be removed for valid numbers
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "007", IntegerType, validatePartitionColumns = false) === "7"
+ )
+
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "00123", LongType, validatePartitionColumns = true) === "123"
+ )
+
+ // Non-numeric types should pass through unchanged
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "abc", StringType, validatePartitionColumns = true) === "abc"
+ )
+
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ "2024-01-15", DateType, validatePartitionColumns = false) ===
"2024-01-15"
+ )
+
+ // Null values should be handled
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ null, IntegerType, validatePartitionColumns = true) === null
+ )
+
+ assert(
+ PartitioningUtils.removeLeadingZerosFromNumberTypePartition(
+ null, StringType, validatePartitionColumns = false) === null
+ )
+ }
+}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala
index 2164f4226aa2..c72dbeec4c8a 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetPartitionDiscoverySuite.scala
@@ -1443,6 +1443,17 @@ class ParquetV2PartitionDiscoverySuite extends
ParquetPartitionDiscoverySuite {
assert(s"p_int=${ExternalCatalogUtils.DEFAULT_PARTITION_NAME}" === path)
}
+ test("SPARK-51830: get path fragment of a non-numeric partition value " +
+ "when partition column validation is disabled") {
+ val spec = Map("p_int" -> "partition_value")
+ val schema = new StructType().add("p_int", "int")
+ assert("p_int=partition_value" ===
+ PartitioningUtils.getPathFragment(spec, schema, validatePartitionColumns
= false))
+ intercept[NumberFormatException] {
+ PartitioningUtils.getPathFragment(spec, schema)
+ }
+ }
+
test("read partitioned table - partition key included in Parquet file") {
withTempDir { base =>
for {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
index 8d2418c5e052..13b437dbf7fe 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.connector.{FakeV2Provider,
FakeV2ProviderWithCustomSchema}
import org.apache.spark.sql.execution.datasources.DataSourceUtils
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.SQLConf.PartitionOverwriteMode
+import org.apache.spark.sql.internal.SQLConf.{PartitionOverwriteMode,
StoreAssignmentPolicy}
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.util.Utils
@@ -3120,6 +3120,60 @@ class InsertSuite extends DataSourceTest with
SharedSparkSession {
}
}
}
+
+ test("SPARK-51830: dynamic partition insert into a table with a non-numeric
partition value") {
+ // `spark.sql.legacy.skipTypeValidationOnAlterPartition` lets us register
a partition whose
+ // value does not match the numeric type of the partition column
+ withSQLConf(SQLConf.SKIP_TYPE_VALIDATION_ON_ALTER_PARTITION.key -> "true")
{
+ withTable("t") {
+ sql("CREATE TABLE t(c1 int, p1 int) USING parquet PARTITIONED BY (p1)")
+ sql("ALTER TABLE t ADD PARTITION (p1 = 'legacy_value')")
+ val legacyLocation = new File(spark.sessionState.catalog
+ .getPartition(TableIdentifier("t"), Map("p1" ->
"legacy_value")).location)
+ assert(legacyLocation.getName === "p1=legacy_value")
+
+ // A dynamic partition insert lists all the partitions of the table to
collect the custom
+ // partition locations, which computes the default path of every
existing partition.
+ intercept[NumberFormatException] {
+ sql("INSERT INTO t VALUES (1, 1)")
+ }
+
+ withSQLConf(SQLConf.VALIDATE_PARTITION_COLUMNS.key -> "false") {
+ sql("INSERT INTO t VALUES (1, 1)")
+ }
+ val newLocation = new File(spark.sessionState.catalog
+ .getPartition(TableIdentifier("t"), Map("p1" -> "1")).location)
+ assert(newLocation.getParentFile === legacyLocation.getParentFile)
+ checkAnswer(spark.read.parquet(newLocation.getCanonicalPath), Row(1))
+ }
+ }
+ }
+
+ test("SPARK-51830: static partition insert with a non-numeric partition
value") {
+ // The LEGACY store assignment policy keeps the non-numeric static
partition value out of an
+ // ANSI cast, so the insert reaches the partition metadata refresh of the
write command.
+ withSQLConf(
+ SQLConf.SKIP_TYPE_VALIDATION_ON_ALTER_PARTITION.key -> "true",
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.LEGACY.toString) {
+ withTable("t") {
+ sql("CREATE TABLE t(c1 int, p1 int) USING parquet PARTITIONED BY (p1)")
+
+ // Inserting no rows into a fully specified static partition registers
the partition
+ // from its path fragment, which is computed from the static partition
value.
+ intercept[NumberFormatException] {
+ sql("INSERT INTO t PARTITION (p1 = 'legacy_value') SELECT c1 FROM
VALUES (1) t(c1)" +
+ " WHERE false")
+ }
+
+ withSQLConf(SQLConf.VALIDATE_PARTITION_COLUMNS.key -> "false") {
+ sql("INSERT INTO t PARTITION (p1 = 'legacy_value') SELECT c1 FROM
VALUES (1) t(c1)" +
+ " WHERE false")
+ }
+
assert(spark.sessionState.catalog.listPartitions(TableIdentifier("t")).map(_.spec)
===
+ Seq(Map("p1" -> "legacy_value")))
+ }
+ }
+ }
}
class FileExistingTestFileSystem extends RawLocalFileSystem {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
