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 1 commit 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
1 commit
Select commit
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
53 changes: 53 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,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) {

@JingsongLi JingsongLi Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve positional mapping when nested field counts differ

Returning input here falls back to the downstream by-name resolver whenever merge-schema fills a missing nested field. For target ARRAY<STRUCT<x: INT, y: INT, z: INT>>, a column-list write whose input struct fields are (y=20, x=10) stores [10, 20, null] instead of the positional [20, 10, null]. I reproduced this on Spark 3.5 with spark.paimon.write.merge-schema=true. Please rename the common ordinal prefix even when the lengths differ, preserve unmatched input fields, and let the existing strict/merge-schema handling process missing or extra fields. A regression test with unequal nested field counts would cover this path.

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"
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
47 changes: 47 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,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<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: 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