[
https://issues.apache.org/jira/browse/SPARK-58816?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18105644#comment-18105644
]
AnhTris commented on SPARK-58816:
---------------------------------
h3. Technical Investigation & Analysis
I have investigated this issue and verified the unexpected behavior on Apache
Spark (tested and reproduced on PySpark 4.0.3 / current master). Below are the
self-contained reproduction steps, the identified root cause in the Catalyst
Analyzer, and a proposed fix direction.
----
h3. Reproduction
The issue can be reproduced directly via PySpark / Spark SQL:
{code:python}
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("SPARK-58816-Reproduction").getOrCreate()
# 1. Create a target table with direct struct, array of structs, and map of
structs
spark.sql("DROP TABLE IF EXISTS target_nested_test")
spark.sql("""
CREATE TABLE target_nested_test (
s STRUCT<x: INT, y: INT>,
arr ARRAY<STRUCT<x: INT, y: INT>>,
m MAP<STRING, STRUCT<x: INT, y: INT>>
) USING parquet
""")
# 2. Insert data where the source struct fields are ordered ('y', 'x') instead
of target's ('x', 'y')
spark.sql("""
INSERT INTO target_nested_test (s, arr, m)
SELECT
named_struct('y', 20, 'x', 10),
array(named_struct('y', 20, 'x', 10)),
map('k', named_struct('y', 20, 'x', 10))
""")
# 3. Query JSON representation of each field
spark.sql("""
SELECT
to_json(s) AS direct_struct,
to_json(arr[0]) AS array_struct,
to_json(m['k']) AS map_struct
FROM target_nested_test
""").show(truncate=False)
{code}
*Actual Output (verified on Spark 4.0.3):*
{noformat}
+---------------+---------------+---------------+
|direct_struct |array_struct |map_struct |
+---------------+---------------+---------------+
|{"x":20,"y":10}|{"x":10,"y":20}|{"x":10,"y":20}|
+---------------+---------------+---------------+
{noformat}
* {{direct_struct}}: {{"x":20,"y":10}} _(Resolved by position: value 20 is
assigned to x, 10 to y)_
* {{array_struct}}: {{"x":10,"y":20}} _(Resolved by name: field 'x' took value
10, 'y' took 20)_
* {{map_struct}}: {{"x":10,"y":20}} _(Resolved by name: field 'x' took value
10, 'y' took 20)_
*Expected Output:*
Under {{INSERT INTO target (column_list)}}, all nested structs should adhere to
SQL positional resolution semantics:
* All three structs must output: {{"x":20,"y":10}}.
----
h3. Root Cause
In Catalyst Analyzer:
# When analyzing {{INSERT INTO table (column_list)}},
{{ResolveInsertionBase.createProjectForByNameQuery}} builds a projection plan
to align input columns to target columns positionally by renaming fields.
# In
{{sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveInsertionBase.scala}}:
#* {{createProjectForByNameQuery}} only pattern matches {{case (input:
StructType, expected: StructType)}}.
#* The helper method {{renameFieldsInStruct}} only accepts {{StructType}} and
recurses through {{StructType}}.
# Because {{ArrayType}} and {{MapType}} are not handled, struct elements within
collections fall through to {{case _ => Alias(queryOutputCol,
resolvedCol.name)()}} and retain their original source field names ('y', 'x').
# Subsequently, when {{TableOutputResolver}} resolves output columns, it falls
back to by-name field resolution for nested collections, silently assigning
values into mismatched fields based on name instead of ordinal position.
----
h3. Proposed Fix
We can generalize {{renameFieldsInStruct}} into {{renameFieldsInDataType}} in
{{ResolveInsertionBase.scala}} to recursively traverse nested collection types
({{ArrayType}} and {{MapType}}):
{code:scala}
// In ResolveInsertionBase.createProjectForByNameQuery:
val projectByName = i.userSpecifiedCols.zip(i.query.output)
.map { case (userSpecifiedCol, queryOutputCol) =>
val resolvedCol = i.table.resolve(Seq(userSpecifiedCol), resolver)
.getOrElse(
throw QueryCompilationErrors.unresolvedAttributeError(
"UNRESOLVED_COLUMN", userSpecifiedCol, i.table.output.map(_.name),
i.origin))
val renamedType = renameFieldsInDataType(queryOutputCol.dataType,
resolvedCol.dataType)
if (queryOutputCol.dataType != renamedType) {
Alias(Cast(queryOutputCol, renamedType), resolvedCol.name)()
} else {
Alias(queryOutputCol, resolvedCol.name)()
}
}
private def renameFieldsInDataType(input: DataType, expected: DataType):
DataType = {
(input, expected) match {
// 1. Recurse through StructType
case (s1: StructType, s2: StructType) if s1.length == s2.length =>
val newFields = s1.zip(s2).map { case (f1, f2) =>
f1.copy(
name = f2.name,
dataType = renameFieldsInDataType(f1.dataType, f2.dataType)
)
}
StructType(newFields)
// 2. Add recursion for ArrayType element types
case (ArrayType(e1, containsNull), ArrayType(e2, _)) =>
ArrayType(renameFieldsInDataType(e1, e2), containsNull)
// 3. Add recursion for MapType key and value types
case (MapType(k1, v1, valContainsNull), MapType(k2, v2, _)) =>
MapType(
renameFieldsInDataType(k1, k2),
renameFieldsInDataType(v1, v2),
valContainsNull
)
// 4. Base case for primitive types
case _ => input
}
}
{code}
This ensures positional rename projection is propagated across all nesting
depths (e.g. {{ARRAY<STRUCT>}}, {{MAP<KEY, STRUCT>}}, {{ARRAY<MAP<...,
STRUCT>>}}), making resolution semantics consistent throughout the entire
schema.
> INSERT with a column list resolves structs inside arrays and maps by name
> instead of position
> ---------------------------------------------------------------------------------------------
>
> Key: SPARK-58816
> URL: https://issues.apache.org/jira/browse/SPARK-58816
> Project: Spark
> Issue Type: Bug
> Components: Bug
> Affects Versions: 3.3.0
> Reporter: Eames Trinh
> Priority: Major
>
> h3. Problem
> For {{{}INSERT INTO table (column_list){}}}, nested structs should be
> resolved by position.
> Spark currently behaves inconsistently:
> * Direct structs are resolved by position.
> * Structs inside arrays and maps are resolved by name.
> This can silently write values into different fields depending only on where
> the struct is nested.
> h3. Reproduction
> {code:java}
> CREATE TABLE target (
> s STRUCT<x: INT, y: INT>,
> arr ARRAY<STRUCT<x: INT, y: INT>>,
> m MAP<STRING, STRUCT<x: INT, y: INT>>
> ) USING parquet;
> INSERT INTO target (s, arr, m)
> SELECT
> named_struct('y', 20, 'x', 10),
> array(named_struct('y', 20, 'x', 10)),
> map('k', named_struct('y', 20, 'x', 10));
> SELECT
> to_json(s) AS direct_struct,
> to_json(arr[0]) AS array_struct,
> to_json(m['k']) AS map_struct
> FROM target; {code}
> h3. Expected result
> All three structs should be resolved by position: {{{"x":20,"y":10}}}
> h3. Actual result
> Only the direct struct is resolved by position. For the array and map structs
> we get: {{{"x":10,"y":20}}}
> h2. Root cause
> {{ResolveInsertionBase.createProjectForByNameQuery}} adds a projection that
> renames direct struct fields according to their positions. However,
> {{renameFieldsInStruct}} only recurses through {{{}StructType{}}}; it does
> not descend into {{ArrayType}} or {{{}MapType{}}}.
> Consequently, field names inside arrays and maps remain unchanged, and
> {{TableOutputResolver}} later resolves those nested structs by name.
> h2. Proposed behavior
> For {{{}INSERT INTO table (column_list){}}}, struct fields should be resolved
> positionally at every nesting level, including:
> * Array elements
> * Map keys
> * Map values
> The projection’s field-renaming logic should recursively traverse arrays and
> maps, or the resolver should otherwise preserve positional semantics
> throughout the nested type.
> h2. Impact
> Queries can silently write values into the wrong nested fields when source
> and target struct field names differ in order. The behavior is also
> inconsistent with direct structs and with the positional semantics of
> {{INSERT INTO}} column lists.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
