From 80c7dce8b4567a940b77f229f76206a82dd8e2eb Mon Sep 17 00:00:00 2001 From: Arnav Balyan Date: Wed, 26 Aug 2026 15:30:22 +0530 Subject: [PATCH] update --- .../catalyst/analysis/PaimonAnalysis.scala | 20 ++++++- .../analysis/PaimonOutputResolver.scala | 53 +++++++++++++++++++ ...stractPaimonSparkSqlExtensionsParser.scala | 19 ++++++- .../sql/InsertOverwriteTableTestBase.scala | 47 ++++++++++++++++ 4 files changed, 137 insertions(+), 2 deletions(-) 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 40276aca1c7f..778a0f0f894b 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 @@ -110,7 +110,16 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { table: DataSourceV2Relation, options: Options, mergeSchemaEnabled: Boolean): LogicalPlan = { - val query = stripHiveDynamicPartitionMarker(v2WriteCommand.query) + val queryWithoutMarker = stripHiveDynamicPartitionMarker(v2WriteCommand.query) + val query = + if ( + v2WriteCommand.isByName && + containsColumnListWrite(v2WriteCommand.query) + ) { + PaimonOutputResolver.renameNestedFieldsByPosition(queryWithoutMarker, table.output) + } else { + queryWithoutMarker + } val hiveStyleDynamicPartitionEnabled = OptionUtils.hiveStyleDynamicPartitionEnabled() hiveDynamicPartitionColumns(v2WriteCommand.query) match { case Some(dynamicPartitionColumns) @@ -154,6 +163,14 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { query.transformDown { case PaimonHiveDynamicPartitionQuery(_, child) => child } } + private def containsColumnListWrite(query: LogicalPlan): Boolean = { + query + .collectFirst { + case node if node.getTagValue(COLUMN_LIST_WRITE).isDefined => true + } + .contains(true) + } + private def resolveDynamicPartitionWrite( query: LogicalPlan, table: DataSourceV2Relation, @@ -273,6 +290,7 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] { object PaimonAnalysis { val PAIMON_WRITE_RESOLVED: TreeNodeTag[Unit] = TreeNodeTag[Unit]("paimon.write.resolved") + val COLUMN_LIST_WRITE: TreeNodeTag[Unit] = TreeNodeTag[Unit]("paimon.write.columnList") } case class PaimonPostHocResolutionRules(session: SparkSession) extends Rule[LogicalPlan] { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala index 37ffebe8280f..2710ccac23ca 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala @@ -58,6 +58,29 @@ object PaimonOutputResolver extends SQLConfHelper { import MissingFieldBehavior._ + def renameNestedFieldsByPosition(query: LogicalPlan, expected: Seq[Attribute]): LogicalPlan = { + val renamed = query.output.map { + input => + expected.find(target => conf.resolver(input.name, target.name)) match { + case Some(target) => + val targetType = + CharVarcharUtils.getRawType(target.metadata).getOrElse(target.dataType) + val renamedType = renameFieldsInType(input.dataType, targetType) + if (renamedType == input.dataType) { + input + } else { + applyColumnMetadata(addCast(input, renamedType), input) + } + case None => input + } + } + if (renamed == query.output) { + query + } else { + Project(renamed, query) + } + } + def resolveOutputColumns( tableName: String, expected: Seq[Attribute], @@ -478,6 +501,36 @@ object PaimonOutputResolver extends SQLConfHelper { attr.withDataType(CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType)) } + private def renameFieldsInStruct(input: StructType, expected: StructType): StructType = { + if (input.length == expected.length) { + StructType(input.zip(expected).map { + case (inputField, expectedField) => + inputField.copy( + name = expectedField.name, + dataType = renameFieldsInType(inputField.dataType, expectedField.dataType)) + }) + } else { + input + } + } + + private def renameFieldsInType(input: DataType, expected: DataType): DataType = { + (input, expected) match { + case (inputStruct: StructType, expectedStruct: StructType) => + renameFieldsInStruct(inputStruct, expectedStruct) + case (ArrayType(inputElement, containsNull), ArrayType(expectedElement, _)) => + ArrayType(renameFieldsInType(inputElement, expectedElement), containsNull) + case ( + MapType(inputKey, inputValue, valueContainsNull), + MapType(expectedKey, expectedValue, _)) => + MapType( + renameFieldsInType(inputKey, expectedKey), + renameFieldsInType(inputValue, expectedValue), + valueContainsNull) + case _ => input + } + } + // Inlined `CharVarcharUtils.CHAR_VARCHAR_TYPE_STRING_METADATA_KEY` — the constant is // `private[sql]` but stable across Spark 3.2–4.1. private val CHAR_VARCHAR_TYPE_STRING_KEY = "__CHAR_VARCHAR_TYPE_STRING" diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala index 7770143c8012..e047498db5a6 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala @@ -19,6 +19,7 @@ package org.apache.spark.sql.catalyst.parser.extensions import org.apache.paimon.spark.SparkProcedures +import org.apache.paimon.spark.catalyst.analysis.PaimonAnalysis.COLUMN_LIST_WRITE import org.apache.paimon.spark.catalyst.plans.logical.PaimonHiveDynamicPartitionQuery import org.antlr.v4.runtime._ @@ -112,6 +113,7 @@ abstract class AbstractPaimonSparkSqlExtensionsParser(val delegate: ParserInterf private def parserRules(sparkSession: SparkSession): Seq[Rule[LogicalPlan]] = { Seq( + MarkInsertColumnList, MarkHiveDynamicPartitionWrite, RewritePaimonViewCommands(sparkSession), RewritePaimonFunctionCommands(sparkSession), @@ -370,6 +372,21 @@ class UpperCaseCharStream(wrapped: CodePointCharStream) extends CharStream { // scalastyle:on } +object MarkInsertColumnList extends Rule[LogicalPlan] { + + override def apply(plan: LogicalPlan): LogicalPlan = { + AnalysisHelper.allowInvokingTransformsInAnalyzer { + plan.transformDown { + case insert: InsertIntoStatement + if insert.userSpecifiedCols.nonEmpty && + !MarkHiveDynamicPartitionWrite.isByName(insert) => + insert.query.setTagValue(COLUMN_LIST_WRITE, ()) + insert + } + } + } +} + object MarkHiveDynamicPartitionWrite extends Rule[LogicalPlan] { override def apply(plan: LogicalPlan): LogicalPlan = { @@ -391,7 +408,7 @@ object MarkHiveDynamicPartitionWrite extends Rule[LogicalPlan] { insert.withNewChildren(Seq(query)).asInstanceOf[InsertIntoStatement] } - private def isByName(insert: InsertIntoStatement): Boolean = { + private[extensions] def isByName(insert: InsertIntoStatement): Boolean = { try { insert.getClass.getMethod("byName").invoke(insert).asInstanceOf[Boolean] } catch { diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala index 40f5d1e120db..ffd8d8da7022 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala @@ -884,4 +884,51 @@ abstract class InsertOverwriteTableTestBase extends PaimonSparkTestBase { .saveAsTable("badTable") }.getMessage.contains("Not a supported type: void")) } + + test("Paimon Insert: column list resolves nested structs positionally") { + withTable("t") { + sql(""" + |CREATE TABLE t ( + | s STRUCT, + | arr ARRAY>, + | map_value MAP>, + | map_key MAP, STRING>, + | deep ARRAY>>> + |) + |""".stripMargin) + + sql(""" + |INSERT INTO t (s, arr, map_value, map_key, deep) + |SELECT + | named_struct('y', 20, 'x', 10), + | array(named_struct('y', 20, 'x', 10)), + | map('k', named_struct('y', 20, 'x', 10)), + | map(named_struct('y', 20, 'x', 10), 'v'), + | array(named_struct( + | 'nested', array(named_struct('y', 20, 'x', 10)))) + |""".stripMargin) + + checkAnswer( + sql("SELECT * FROM t"), + Row( + Row(20, 10), + Seq(Row(20, 10)), + Map("k" -> Row(20, 10)), + Map(Row(20, 10) -> "v"), + Seq(Row(Seq(Row(20, 10)))))) + } + } + + test("Paimon Insert: by name resolves nested structs by name") { + if (gteqSpark3_5) { + withTable("t") { + sql("CREATE TABLE t (arr ARRAY>)") + sql(""" + |INSERT INTO t BY NAME + |SELECT array(named_struct('y', 20, 'x', 10)) AS arr + |""".stripMargin) + checkAnswer(sql("SELECT * FROM t"), Row(Seq(Row(10, 20)))) + } + } + } }