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 45999fe254 [spark] Support LEGACY store assignment for format tables
(#10013)
45999fe254 is described below
commit 45999fe254e5f5467f8598ec103e49c511951896
Author: Zouxxyy <[email protected]>
AuthorDate: Mon Sep 21 10:40:59 2026 +0800
[spark] Support LEGACY store assignment for format tables (#10013)
---
.../paimon/spark/catalyst/Compatibility.scala | 23 +-
.../paimon/spark/catalyst/Compatibility.scala | 12 +-
.../paimon/spark/catalyst/Compatibility.scala | 12 +-
.../paimon/spark/catalyst/Compatibility.scala | 18 +-
.../spark/catalyst/analysis/PaimonAnalysis.scala | 3 +
.../analysis/PaimonFormatOutputResolver.scala | 59 +++++
.../paimon/spark/format/PaimonFormatTable.scala | 21 +-
.../paimon/spark/sql/FormatTableTestBase.scala | 272 ++++++++++++++++++++-
8 files changed, 410 insertions(+), 10 deletions(-)
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 619521c63c..b7c1e31f0f 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 619521c63c..4703e43663 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-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
similarity index 81%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
copy to
paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/paimon/spark/catalyst/Compatibility.scala
index 443024058e..2c3dbbf06f 100644
---
a/paimon-spark/paimon-spark-common/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
@@ -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 443024058e..7222e0a27c 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 6353422342..527ca1faad 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 0000000000..b640e2917f
--- /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 c610b9d031..90fdd7a988 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 a7e3bdce5e..bbcd1cc899 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<id: BIGINT>, a ARRAY<BIGINT>, m
MAP<STRING, BIGINT>) " +
+ "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' ';')")
