Skip to content

[spark] Fix nested struct values wrong order for insert columns #9415

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ArnavBalyan wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[spark] Fix nested struct values wrong order for insert columns #9415

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
80c7dce
update
ArnavBalyan Aug 26, 2026
d3c149f
update
ArnavBalyan Aug 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 19 additions & 1 deletion ...park-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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] {
Expand Down
51 changes: 51 additions & 0 deletions ...ommon/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonOutputResolver.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand Down Expand Up @@ -478,6 +501,34 @@ object PaimonOutputResolver extends SQLConfHelper {
attr.withDataType(CharVarcharUtils.getRawType(attr.metadata).getOrElse(attr.dataType))
}

private def renameFieldsInStruct(input: StructType, expected: StructType): StructType = {
StructType(input.zipWithIndex.map {
case (inputField, index) if index < expected.length =>
val expectedField = expected(index)
inputField.copy(
name = expectedField.name,
dataType = renameFieldsInType(inputField.dataType, expectedField.dataType))
case (inputField, _) => inputField
})
}

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"
Expand Down
19 changes: 18 additions & 1 deletion .../apache/spark/sql/catalyst/parser/extensions/AbstractPaimonSparkSqlExtensionsParser.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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._
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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 = {
Expand All @@ -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 {
Expand Down
60 changes: 60 additions & 0 deletions ...on-spark-ut/src/test/scala/org/apache/paimon/spark/sql/InsertOverwriteTableTestBase.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -884,4 +884,64 @@ 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<x: INT, y: INT>,
| arr ARRAY<STRUCT<x: INT, y: INT>>,
| map_value MAP<STRING, STRUCT<x: INT, y: INT>>,
| map_key MAP<STRUCT<x: INT, y: INT>, STRING>,
| deep ARRAY<STRUCT<nested: ARRAY<STRUCT<x: INT, y: INT>>>>
|)
|""".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: column list resolves unequal nested structs positionally") {
withSparkSQLConf("spark.paimon.write.merge-schema" -> "true") {
withTable("t") {
sql("CREATE TABLE t (arr ARRAY<STRUCT<x: INT, y: INT, z: INT>>)")
sql("""
|INSERT INTO t (arr)
|SELECT array(named_struct('y', 20, 'x', 10))
|""".stripMargin)
checkAnswer(sql("SELECT * FROM t"), Row(Seq(Row(20, 10, null))))
}
}
}

test("Paimon Insert: by name resolves nested structs by name") {
if (gteqSpark3_5) {
withTable("t") {
sql("CREATE TABLE t (arr ARRAY<STRUCT<x: INT, y: INT>>)")
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))))
}
}
}
}
Loading