From 630a97748e168bc530eed6c323619d39979bb50d Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Sun, 20 Sep 2026 18:26:40 +0800 Subject: [PATCH 1/2] [spark] Support LEGACY store assignment for format tables --- .../paimon/spark/catalyst/Compatibility.scala | 23 +- .../paimon/spark/catalyst/Compatibility.scala | 12 +- .../paimon/spark/catalyst/Compatibility.scala | 18 +- .../catalyst/analysis/PaimonAnalysis.scala | 3 + .../analysis/PaimonFormatOutputResolver.scala | 59 ++++ .../spark/format/PaimonFormatTable.scala | 21 +- .../spark/sql/FormatTableTestBase.scala | 272 +++++++++++++++++- 7 files changed, 399 insertions(+), 9 deletions(-) create mode 100644 paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFormatOutputResolver.scala diff --git a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala index 619521c63c58..b7c1e31f0f00 100644 --- a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala +++ b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala @@ -19,7 +19,8 @@ package org.apache.paimon.spark.catalyst import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.expressions.{Cast, Expression} +import org.apache.spark.sql.catalyst.analysis.TableOutputResolver +import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression} import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand} import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.execution.ui.SQLPlanMetric @@ -28,6 +29,26 @@ import org.apache.spark.sql.types.DataType object Compatibility { + def resolveTableOutputColumns( + tableName: String, + expected: Seq[Attribute], + query: LogicalPlan, + byName: Boolean, + conf: SQLConf): LogicalPlan = { + // SPARK-38228 fixed this separation in 3.3: LEGACY assignment must use non-ANSI casts even + // when ANSI expression evaluation is enabled. Scope the override to this resolution only. + val assignmentConf = if (conf.storeAssignmentPolicy == SQLConf.StoreAssignmentPolicy.LEGACY) { + val legacyConf = conf.clone() + legacyConf.setConf(SQLConf.ANSI_ENABLED, false) + legacyConf + } else { + conf + } + SQLConf.withExistingConf(assignmentConf) { + TableOutputResolver.resolveOutputColumns(tableName, expected, query, byName, assignmentConf) + } + } + def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = { o.withNewQuery(query) } diff --git a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala index 619521c63c58..4703e436630a 100644 --- a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala +++ b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala @@ -19,7 +19,8 @@ package org.apache.paimon.spark.catalyst import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.expressions.{Cast, Expression} +import org.apache.spark.sql.catalyst.analysis.TableOutputResolver +import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression} import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand} import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.execution.ui.SQLPlanMetric @@ -28,6 +29,15 @@ import org.apache.spark.sql.types.DataType object Compatibility { + def resolveTableOutputColumns( + tableName: String, + expected: Seq[Attribute], + query: LogicalPlan, + byName: Boolean, + conf: SQLConf): LogicalPlan = { + TableOutputResolver.resolveOutputColumns(tableName, expected, query, byName, conf) + } + def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = { o.withNewQuery(query) } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala index 443024058ed4..7222e0a27c68 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala @@ -19,7 +19,8 @@ package org.apache.paimon.spark.catalyst import org.apache.spark.sql.SparkSession -import org.apache.spark.sql.catalyst.expressions.{Cast, Expression} +import org.apache.spark.sql.catalyst.analysis.TableOutputResolver +import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression} import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand} import org.apache.spark.sql.catalyst.trees.TreeNodeTag import org.apache.spark.sql.execution.ui.SQLPlanMetric @@ -28,6 +29,21 @@ import org.apache.spark.sql.types.DataType object Compatibility { + def resolveTableOutputColumns( + tableName: String, + expected: Seq[Attribute], + query: LogicalPlan, + byName: Boolean, + conf: SQLConf): LogicalPlan = { + TableOutputResolver.resolveOutputColumns( + tableName, + expected, + query, + byName, + conf, + supportColDefaultValue = false) + } + def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = { o.withNewQuery(query) } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala index 63534223420f..527ca1faad64 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala @@ -52,6 +52,9 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { case o @ PaimonDynamicPartitionOverwrite(r, d) if o.resolved => PaimonDynamicPartitionOverwriteCommand(r, d, o.query, o.writeOptions, o.isByName) + case a: V2WriteCommand if PaimonFormatOutputResolver.isLegacyFormatWrite(a) => + PaimonFormatOutputResolver.resolve(a) + case a @ PaimonV2WriteCommand(table) if a.query.getTagValue(PAIMON_WRITE_RESOLVED).isEmpty => val options = Options.fromMap(writeOptions(a).asJava) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFormatOutputResolver.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFormatOutputResolver.scala new file mode 100644 index 000000000000..b640e2917ff9 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonFormatOutputResolver.scala @@ -0,0 +1,59 @@ +/* + * 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.catalyst.analysis + +import org.apache.paimon.spark.catalyst.Compatibility +import org.apache.paimon.spark.format.PaimonFormatTable + +import org.apache.spark.sql.catalyst.SQLConfHelper +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand} +import org.apache.spark.sql.catalyst.trees.TreeNodeTag +import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation +import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy + +/** Align LEGACY format writes without enabling missing-column filling or schema evolution. */ +object PaimonFormatOutputResolver extends SQLConfHelper { + + private val resolvedQuery = TreeNodeTag[LogicalPlan]("paimon.format.write.resolved-query") + + def isLegacyFormatWrite(write: V2WriteCommand): Boolean = write.table match { + case relation: DataSourceV2Relation if relation.table.isInstanceOf[PaimonFormatTable] => + write.query.resolved && conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY + case _ => false + } + + def resolve(write: V2WriteCommand): V2WriteCommand = { + if (write.getTagValue(resolvedQuery).contains(write.query)) { + write + } else { + val query = Compatibility.resolveTableOutputColumns( + write.table.name, + write.table.output, + write.query, + write.isByName, + conf) + val resolved = Compatibility.withNewQuery(write, query) + // LEGACY casts may remain nullable for NOT NULL targets. Keep the existing writer-side + // nullability checks; do not repeatedly align the query or falsify expression nullability. + // Tag the command rather than its source query, which can be reused for another table. + resolved.setTagValue(resolvedQuery, query) + resolved + } + } +} 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 c610b9d031c9..90fdd7a9888d 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 @@ -34,14 +34,15 @@ 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.catalog.TableCapability.{ACCEPT_ANY_SCHEMA, BATCH_READ, BATCH_WRITE, OVERWRITE_BY_FILTER, OVERWRITE_DYNAMIC} import org.apache.spark.sql.connector.distributions.Distribution import org.apache.spark.sql.connector.expressions.SortOrder 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.internal.SQLConf import org.apache.spark.sql.paimon.shims.SparkShimLoader -import org.apache.spark.sql.types.{StringType, StructType} +import org.apache.spark.sql.types.{DataType, StringType, StructType} import org.apache.spark.sql.util.CaseInsensitiveStringMap import java.util @@ -71,7 +72,14 @@ case class PaimonFormatTable(table: FormatTable) def hasCatalogManagedPartitions: Boolean = partitionManager != null override def capabilities(): util.Set[TableCapability] = { - util.EnumSet.of(BATCH_READ, BATCH_WRITE, OVERWRITE_DYNAMIC, OVERWRITE_BY_FILTER) + val capabilities = + util.EnumSet.of(BATCH_READ, BATCH_WRITE, OVERWRITE_DYNAMIC, OVERWRITE_BY_FILTER) + // Only LEGACY needs our output resolver: Spark's generic V2 analyzer rejects it before + // aligning columns. Leave ANSI/STRICT on Spark's path, also when our extension is not loaded. + if (SQLConf.get.storeAssignmentPolicy == SQLConf.StoreAssignmentPolicy.LEGACY) { + capabilities.add(ACCEPT_ANY_SCHEMA) + } + capabilities } override def properties: util.Map[String, String] = { @@ -97,6 +105,13 @@ case class PaimonFormatTable(table: FormatTable) } override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = { + require( + DataType.equalsIgnoreNullability(schema, info.schema), + "Format table writes must match the table schema. " + + s"Expected ${schema.catalogString}, but found ${info.schema.catalogString}. " + + "Configure org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions " + + "to resolve write columns and types." + ) PaimonFormatTableWriterBuilder(table, info.schema) } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala index a7e3bdce5e0f..bbcd1cc89920 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/FormatTableTestBase.scala @@ -18,21 +18,32 @@ package org.apache.paimon.spark.sql -import org.apache.paimon.catalog.{DelegateCatalog, Identifier} +import org.apache.paimon.catalog.{CatalogContext, DelegateCatalog, Identifier} import org.apache.paimon.fs.Path import org.apache.paimon.hive.HiveCatalog +import org.apache.paimon.options.Options import org.apache.paimon.spark.{PaimonFormatTableScan, PaimonHiveTestBase, PaimonInputPartition} import org.apache.paimon.spark.PaimonHiveTestBase.hiveUri +import org.apache.paimon.spark.format.PaimonFormatTable import org.apache.paimon.table.FormatTable import org.apache.paimon.table.source.Split +import org.apache.paimon.types.DataTypes import org.apache.paimon.utils.{CompressUtils, PartitionPathUtils} -import org.apache.spark.sql.Row +import org.apache.spark.sql.{AnalysisException, PaimonUtils, Row} import org.apache.spark.sql.catalyst.analysis.NoSuchPartitionException +import org.apache.spark.sql.catalyst.plans.logical.AppendData +import org.apache.spark.sql.connector.catalog.{Identifier => SparkIdentifier, TableCatalog} import org.apache.spark.sql.connector.read.InputPartition +import org.apache.spark.sql.connector.write.LogicalWriteInfo import org.apache.spark.sql.execution.SparkPlan import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper -import org.apache.spark.sql.execution.datasources.v2.BatchScanExec +import org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand +import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DataSourceV2Relation, FileTable} +import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType} +import org.apache.spark.sql.util.CaseInsensitiveStringMap + +import java.util.{Collections, Locale} abstract class FormatTableTestBase extends PaimonHiveTestBase with AdaptiveSparkPlanHelper { @@ -43,6 +54,261 @@ abstract class FormatTableTestBase extends PaimonHiveTestBase with AdaptiveSpark sql(s"USE $hiveDbName") } + test("Format table: LEGACY casts strings without changing the SQL or schema") { + for (ansiEnabled <- Seq("true", "false")) { + withSparkSQLConf( + "spark.sql.storeAssignmentPolicy" -> "LEGACY", + "spark.sql.ansi.enabled" -> ansiEnabled, + "spark.paimon.format-table.implementation" -> "paimon") { + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT, label STRING) USING parquet") + val schema = spark.table("t").schema + sql("INSERT INTO t SELECT '123', 'valid'") + sql( + "INSERT INTO t VALUES ('-456', 'negative'), ('bad', 'invalid'), " + + "('9223372036854775808', 'overflow'), (NULL, 'null')") + checkAnswer( + sql("SELECT * FROM t"), + Seq( + Row(123L, "valid"), + Row(-456L, "negative"), + Row(null, "invalid"), + Row(null, "overflow"), + Row(null, "null"))) + assert(spark.table("t").schema == schema) + assert(spark.conf.get("spark.sql.ansi.enabled") == ansiEnabled) + } + } + } + } + + test("Format table: LEGACY aligns by name without filling missing columns or evolving schema") { + withSparkSQLConf( + "spark.sql.storeAssignmentPolicy" -> "LEGACY", + "spark.paimon.format-table.implementation" -> "paimon", + "spark.paimon.write.merge-schema" -> "true") { + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT, label STRING) USING parquet") + val schema = spark.table("t").schema + val source = Seq(("first", "123")).toDF("label", "group_id") + source.writeTo("t").append() + // Reusing the source for a different table must not reuse that table's alignment. + withTable("other") { + sql("CREATE TABLE other (group_id STRING, label STRING) USING parquet") + source.writeTo("other").append() + checkAnswer(sql("SELECT * FROM other"), Row("123", "first")) + } + intercept[AnalysisException] { + source.drop("label").writeTo("t").append() + } + intercept[AnalysisException] { + Seq(("123", "first", "extra")) + .toDF("group_id", "label", "extra") + .writeTo("t") + .append() + } + intercept[AnalysisException] { + Seq(("123", "first")).toDF("wrong_name", "label").writeTo("t").append() + } + intercept[AnalysisException] { + sql("INSERT INTO t SELECT '123'") + } + intercept[AnalysisException] { + sql("INSERT INTO t SELECT '123', 'first', 'extra'") + } + assert(spark.table("t").schema == schema) + checkAnswer(sql("SELECT * FROM t"), Row(123L, "first")) + } + } + } + + test("Format table: LEGACY keeps fixed nested schemas and Spark's field matching rules") { + withSparkSQLConf( + "spark.sql.storeAssignmentPolicy" -> "LEGACY", + "spark.paimon.format-table.implementation" -> "paimon") { + withTable("t") { + sql( + "CREATE TABLE t (s STRUCT, a ARRAY, m MAP) " + + "USING parquet") + sql("INSERT INTO t SELECT named_struct('id', '123'), array('4', '5'), map('key', '6')") + checkAnswer(sql("SELECT * FROM t"), Row(Row(123L), Seq(4L, 5L), Map("key" -> 6L))) + intercept[AnalysisException] { + sql( + "INSERT INTO t SELECT named_struct('id', '1', 'extra', '2'), " + + "array('4'), map('key', '6')") + } + val renamedField = + sql("SELECT named_struct('other', '1') AS s, array('4') AS a, map('key', '6') AS m") + if (gteqSpark3_3) { + intercept[AnalysisException] { + renamedField.writeTo("t").append() + } + } else { + // Spark 3.2's resolver aligns nested structs positionally, even for a by-name write. + // Preserve its matching semantics while retaining the target's field names and types. + renamedField.writeTo("t").append() + checkAnswer( + sql("SELECT * FROM t"), + Seq( + Row(Row(123L), Seq(4L, 5L), Map("key" -> 6L)), + Row(Row(1L), Seq(4L), Map("key" -> 6L)))) + } + } + } + } + + test("Format table: LEGACY casts append and static or dynamic partition overwrites") { + withSparkSQLConf( + "spark.sql.storeAssignmentPolicy" -> "LEGACY", + "spark.paimon.format-table.implementation" -> "paimon") { + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT, pt INT) USING parquet PARTITIONED BY (pt)") + sql("INSERT INTO t VALUES ('1', '1'), ('2', '2')") + sql("INSERT OVERWRITE t PARTITION (pt=1) SELECT '3'") + checkAnswer(sql("SELECT * FROM t"), Seq(Row(3L, 1), Row(2L, 2))) + withSparkSQLConf("spark.sql.sources.partitionOverwriteMode" -> "dynamic") { + sql("INSERT OVERWRITE t PARTITION (pt) SELECT '4', '2'") + } + checkAnswer(sql("SELECT * FROM t"), Seq(Row(3L, 1), Row(4L, 2))) + } + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT) USING parquet") + sql("INSERT INTO t SELECT '1'") + sql("INSERT OVERWRITE t SELECT '2'") + checkAnswer(sql("SELECT * FROM t"), Row(2L)) + } + } + } + + test("Format table: ANSI and STRICT retain their assignment rules for both implementations") { + for (implementation <- Seq("paimon", "engine"); policy <- Seq("ANSI", "STRICT")) { + withSparkSQLConf( + "spark.paimon.format-table.implementation" -> implementation, + "spark.sql.storeAssignmentPolicy" -> policy, + "spark.sql.ansi.enabled" -> "false") { + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT) USING parquet") + intercept[AnalysisException] { + sql("INSERT INTO t SELECT '123'") + } + sql("INSERT INTO t SELECT 123") + checkAnswer(sql("SELECT * FROM t"), Row(123L)) + } + if (policy == "STRICT") { + withTable("t") { + sql("CREATE TABLE t (group_id INT) USING parquet") + intercept[AnalysisException] { + sql("INSERT INTO t SELECT 123L") + } + } + } else { + withTable("t") { + sql("CREATE TABLE t (group_id INT) USING parquet") + val error = intercept[Exception] { + sql("INSERT INTO t SELECT 2147483648L") + } + assert( + Iterator + .iterate[Throwable](error)(_.getCause) + .takeWhile(_ != null) + .exists( + e => + Option(e.getMessage).exists(_.toLowerCase(Locale.ROOT).contains("overflow")))) + } + } + } + } + } + + test("Format table: engine retains Spark's V1 write path and LEGACY casts") { + withSparkSQLConf( + "spark.paimon.format-table.implementation" -> "engine", + "spark.sql.storeAssignmentPolicy" -> "LEGACY") { + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT) USING parquet") + val table = spark.sessionState.catalogManager.currentCatalog + .asInstanceOf[TableCatalog] + .loadTable(SparkIdentifier.of(Array(hiveDbName), "t")) + assert(table.isInstanceOf[FileTable], table.getClass.getName) + assert(table.schema().fields.head.dataType == LongType) + val result = sql("INSERT INTO t SELECT '123'") + assert(result.queryExecution.analyzed.collect { + case command: InsertIntoHadoopFsRelationCommand => command + }.nonEmpty) + sql("INSERT INTO t SELECT 'bad'") + checkAnswer(sql("SELECT * FROM t"), Seq(Row(123L), Row(null))) + } + } + } + + test("Format table: LEGACY casts preserve NOT NULL enforcement and analyzer convergence") { + withSparkSQLConf( + "spark.sql.storeAssignmentPolicy" -> "LEGACY", + "spark.sql.ansi.enabled" -> "true") { + withTempDir { + dir => + // Build the relation directly so the Hive metastore cannot erase field nullability. + val options = new Options + options.setString("path", dir.toURI.toString) + options.setString("file.format", "parquet") + val table = FormatTable + .builder() + .fileIO(fileIO) + .identifier(Identifier.create(hiveDbName, "not_null_format")) + .rowType(DataTypes.ROW(DataTypes.FIELD(0, "group_id", DataTypes.BIGINT().notNull()))) + .partitionKeys(Collections.emptyList[String]()) + .location(dir.toURI.toString) + .format(FormatTable.Format.PARQUET) + .options(options.toMap) + .catalogContext(CatalogContext.create(new Options)) + .build() + val relation = DataSourceV2Relation.create(PaimonFormatTable(table), None, None) + + def insert(value: String): Unit = { + val query = Seq(value).toDF("group_id").queryExecution.analyzed + PaimonUtils.createDataset(spark, AppendData.byPosition(relation, query)).collect() + } + + insert("123") + checkAnswer(PaimonUtils.createDataset(spark, relation), Row(123L)) + val error = intercept[Exception] { + insert("bad") + } + assert( + Iterator + .iterate[Throwable](error)(_.getCause) + .takeWhile(_ != null) + .exists( + e => Option(e.getMessage).exists(_.contains("Cannot write null to non-null")))) + } + } + } + + test("Format table: write builder rejects a schema that bypassed output resolution") { + withTable("t") { + sql("CREATE TABLE t (group_id BIGINT) USING parquet") + val table = + paimonCatalog.getTable(Identifier.create(hiveDbName, "t")).asInstanceOf[FormatTable] + val sparkTable = PaimonFormatTable(table) + for ( + inputSchema <- Seq( + StructType(Seq(StructField("group_id", StringType))), + StructType(Seq(StructField("wrong_name", LongType))), + StructType(Nil)) + ) { + val error = intercept[IllegalArgumentException] { + sparkTable.newWriteBuilder(new LogicalWriteInfo { + override def queryId(): String = "unresolved-format-write" + override def schema(): StructType = inputSchema + override def options(): CaseInsensitiveStringMap = + new CaseInsensitiveStringMap(Collections.emptyMap[String, String]()) + }) + } + assert(error.getMessage.contains("Format table writes must match the table schema")) + } + } + } + test("Format table: csv with field-delimiter") { withTable("t") { sql(s"CREATE TABLE t (f0 INT, f1 INT) USING CSV OPTIONS ('csv.field-delimiter' ';')") From 52c7b346129f165edd0f27eae10bec580e3a22b0 Mon Sep 17 00:00:00 2001 From: zouxxyy Date: Sun, 20 Sep 2026 20:44:32 +0800 Subject: [PATCH 2/2] [spark] Fix format output resolution on Spark 3.4 Spark 3.4 exposes the five-argument TableOutputResolver API. Add its version-specific Compatibility implementation instead of reusing the Spark 3.5 six-argument call, which aborts format writes with a NoSuchMethodError. Validated FormatTableTest on Spark 3.4.3 with Scala 2.12 and Scala 2.13: 33 tests passed for each variant, with standard Maven verify. --- .../paimon/spark/catalyst/Compatibility.scala | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala diff --git a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala new file mode 100644 index 000000000000..2c3dbbf06f75 --- /dev/null +++ b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala @@ -0,0 +1,60 @@ +/* + * 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.catalyst + +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.analysis.TableOutputResolver +import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Expression} +import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, V2WriteCommand} +import org.apache.spark.sql.catalyst.trees.TreeNodeTag +import org.apache.spark.sql.execution.ui.SQLPlanMetric +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.DataType + +object Compatibility { + + def resolveTableOutputColumns( + tableName: String, + expected: Seq[Attribute], + query: LogicalPlan, + byName: Boolean, + conf: SQLConf): LogicalPlan = { + TableOutputResolver.resolveOutputColumns(tableName, expected, query, byName, conf) + } + + def withNewQuery(o: V2WriteCommand, query: LogicalPlan): V2WriteCommand = { + o.withNewQuery(query) + } + + def castByTableInsertionTag: TreeNodeTag[Unit] = { + Cast.BY_TABLE_INSERTION + } + + def cast( + child: Expression, + dataType: DataType, + timeZoneId: Option[String] = None, + ansiEnabled: Boolean = SQLConf.get.ansiEnabled): Cast = { + Cast(child, dataType, timeZoneId, ansiEnabled) + } + + def getExecutionMetrics(spark: SparkSession, executionId: Long): Seq[SQLPlanMetric] = { + spark.sharedState.statusStore.execution(executionId).get.metrics.toSeq + } +}