You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by be...@apache.org on 2025/03/13 02:50:12 UTC
(spark) branch branch-4.0 updated: [SPARK-48922][SQL] Avoid redundant array transform of identical expression for map type
This is an automated email from the ASF dual-hosted git repository.
beliefer pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 7f6ac7f2527d [SPARK-48922][SQL] Avoid redundant array transform of identical expression for map type
7f6ac7f2527d is described below
commit 7f6ac7f2527d71f3659058fb2bb2f2c148347960
Author: wforget <64...@qq.com>
AuthorDate: Thu Mar 13 10:39:05 2025 +0800
[SPARK-48922][SQL] Avoid redundant array transform of identical expression for map type
### What changes were proposed in this pull request?
Similar to #47843, this patch avoids ArrayTransform in `resolveMapType` function if the resolution expression is the same as input param.
### Why are the changes needed?
My previous pr #47381 was not merged, but I still think it is an optimization, so I reopened it.
During the upgrade from Spark 3.1.1 to 3.5.0, I found a performance regression in map type inserts.
There are some extra conversion expressions in project before insert, which doesn't seem to be always necessary.
```
map_from_arrays(transform(map_keys(map#516), lambdafunction(lambda key#652, lambda key#652, false)), transform(map_values(map#516), lambdafunction(lambda value#654, lambda value#654, false))) AS map#656
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
added unit test
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #50245 from wForget/SPARK-48922.
Authored-by: wforget <64...@qq.com>
Signed-off-by: beliefer <be...@163.com>
(cherry picked from commit 1be108eedb832a3684fcc55ec15581a1347475f4)
Signed-off-by: beliefer <be...@163.com>
---
.../catalyst/analysis/TableOutputResolver.scala | 27 +++++++++++++++----
.../catalyst/analysis/V2WriteAnalysisSuite.scala | 31 ++++++++++++++++++++++
.../subquery/subquery-nested-data.sql.out | 4 +--
3 files changed, 55 insertions(+), 7 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
index 4a6504666d41..52af6342c012 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
@@ -460,11 +460,28 @@ object TableOutputResolver extends SQLConfHelper with Logging {
}
if (resKey.length == 1 && resValue.length == 1) {
- val keyFunc = LambdaFunction(resKey.head, Seq(keyParam))
- val valueFunc = LambdaFunction(resValue.head, Seq(valueParam))
- val newKeys = ArrayTransform(MapKeys(nullCheckedInput), keyFunc)
- val newValues = ArrayTransform(MapValues(nullCheckedInput), valueFunc)
- Some(Alias(MapFromArrays(newKeys, newValues), expected.name)())
+ // If the key and value expressions have not changed, we just check original map field.
+ // Otherwise, we construct a new map by adding transformations to the keys and values.
+ if (resKey.head == keyParam && resValue.head == valueParam) {
+ Some(
+ Alias(nullCheckedInput, expected.name)(
+ nonInheritableMetadataKeys =
+ Seq(CharVarcharUtils.CHAR_VARCHAR_TYPE_STRING_METADATA_KEY)))
+ } else {
+ val newKeys = if (resKey.head != keyParam) {
+ val keyFunc = LambdaFunction(resKey.head, Seq(keyParam))
+ ArrayTransform(MapKeys(nullCheckedInput), keyFunc)
+ } else {
+ MapKeys(nullCheckedInput)
+ }
+ val newValues = if (resValue.head != valueParam) {
+ val valueFunc = LambdaFunction(resValue.head, Seq(valueParam))
+ ArrayTransform(MapValues(nullCheckedInput), valueFunc)
+ } else {
+ MapValues(nullCheckedInput)
+ }
+ Some(Alias(MapFromArrays(newKeys, newValues), expected.name)())
+ }
} else {
None
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2WriteAnalysisSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2WriteAnalysisSuite.scala
index 29c6c63ecfea..f2e3a4bf9195 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2WriteAnalysisSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/V2WriteAnalysisSuite.scala
@@ -333,6 +333,37 @@ abstract class V2WriteAnalysisSuiteBase extends AnalysisTest {
hasTransform = true)
}
+ test("SPARK-48922: Avoid redundant array transform of identical expression for map type") {
+ def assertMapField(fromType: MapType, toType: MapType, transformNum: Int): Unit = {
+ val table = TestRelation(Seq($"a".int, Symbol("map").map(toType)))
+ val query = TestRelation(Seq(Symbol("map").map(fromType), $"a".int))
+
+ val writePlan = byName(table, query).analyze
+
+ assertResolved(writePlan)
+ checkAnalysis(writePlan, writePlan)
+
+ val transforms = writePlan.children.head.expressions.flatMap { e =>
+ e.flatMap {
+ case t: ArrayTransform => Some(t)
+ case _ => None
+ }
+ }
+ assert(transforms.size == transformNum)
+ }
+
+ assertMapField(MapType(LongType, StringType), MapType(LongType, StringType), 0)
+ assertMapField(
+ MapType(LongType, new StructType().add("x", "int").add("y", "int")),
+ MapType(LongType, new StructType().add("y", "int").add("x", "byte")),
+ 1)
+ assertMapField(MapType(LongType, LongType), MapType(IntegerType, LongType), 1)
+ assertMapField(
+ MapType(LongType, new StructType().add("x", "int").add("y", "int")),
+ MapType(IntegerType, new StructType().add("y", "int").add("x", "byte")),
+ 2)
+ }
+
test("SPARK-33136: output resolved on complex types for V2 write commands") {
def assertTypeCompatibility(name: String, fromType: DataType, toType: DataType): Unit = {
val table = TestRelation(StructType(Seq(StructField("a", toType))))
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/subquery/subquery-nested-data.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/subquery/subquery-nested-data.sql.out
index 9cbd10e4d2ca..a218238073df 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/subquery/subquery-nested-data.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/subquery/subquery-nested-data.sql.out
@@ -23,7 +23,7 @@ CreateDataSourceTableCommand `spark_catalog`.`default`.`x`, false
insert into x values (map(1, 2), 3), (map(1, 4), 5), (map(2, 3), 4), (map(5, 6), 7)
-- !query analysis
InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/x, false, Parquet, [path=file:[not included in comparison]/{warehouse_dir}/x], Append, `spark_catalog`.`default`.`x`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/x), [xm, x2]
-+- Project [map_from_arrays(transform(map_keys(col1#x), lambdafunction(lambda key#x, lambda key#x, false)), transform(map_values(col1#x), lambdafunction(lambda value#x, lambda value#x, false))) AS xm#x, cast(col2#x as int) AS x2#x]
++- Project [col1#x AS xm#x, cast(col2#x as int) AS x2#x]
+- LocalRelation [col1#x, col2#x]
@@ -37,7 +37,7 @@ CreateDataSourceTableCommand `spark_catalog`.`default`.`y`, false
insert into y values (map(1, 2), 10), (map(1, 3), 20), (map(2, 3), 20), (map(8, 3), 20)
-- !query analysis
InsertIntoHadoopFsRelationCommand file:[not included in comparison]/{warehouse_dir}/y, false, Parquet, [path=file:[not included in comparison]/{warehouse_dir}/y], Append, `spark_catalog`.`default`.`y`, org.apache.spark.sql.execution.datasources.InMemoryFileIndex(file:[not included in comparison]/{warehouse_dir}/y), [ym, y2]
-+- Project [map_from_arrays(transform(map_keys(col1#x), lambdafunction(lambda key#x, lambda key#x, false)), transform(map_values(col1#x), lambdafunction(lambda value#x, lambda value#x, false))) AS ym#x, cast(col2#x as int) AS y2#x]
++- Project [col1#x AS ym#x, cast(col2#x as int) AS y2#x]
+- LocalRelation [col1#x, col2#x]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org