You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2025/06/23 12:27:20 UTC
(spark) branch master updated: [SPARK-52462][SQL] Enforce type coercion before children output deduplication in Union
This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b5948486cbd5 [SPARK-52462][SQL] Enforce type coercion before children output deduplication in Union
b5948486cbd5 is described below
commit b5948486cbd517040c6a0d674af220331175bfca
Author: mihailoale-db <mi...@databricks.com>
AuthorDate: Mon Jun 23 20:27:05 2025 +0800
[SPARK-52462][SQL] Enforce type coercion before children output deduplication in Union
### What changes were proposed in this pull request?
Right now, query the following query produces plans that are not consistent over different underlying table providers.
Mentioned can happen when introducing some third party data sources that may add custom analyzer rules that will change the rule order here. Delta Lake is an example.
See these examples:
- Delta table:
```
CREATE TABLE deltaTable (col1 INT, col2 INT, col3 INT, col4 INT) USING delta;
SELECT col1, col2, col3, NULLIF('','') AS col4
FROM deltaTable
UNION ALL
SELECT col2, col2, null AS col3, col4
FROM deltaTable;
```
For this one, we trigger `WidenSetOperationTypes` before `ResolveReferences` (deduplication of `Union` children outputs) and thus plan looks like:
```
Union false, false
:- Project [col1#418, col2#419, col3#420, cast(col4#412 as bigint) AS col4#426L]
: +- Project [col1#418, col2#419, col3#420, nullif(, ) AS col4#412]
: +- SubqueryAlias spark_catalog.default.deltaTable
: +- Relation spark_catalog.default.deltatable[col1#418,col2#419,col3#420,col4#421] parquet
+- Project [col2#423, col2#423 AS col2#431, col3#427, col4#428L]
+- Project [col2#423, col2#423, cast(col3#413 as int) AS col3#427, cast(col4#425 as bigint) AS col4#428L]
+- Project [col2#423, col2#423, null AS col3#413, col4#425]
+- SubqueryAlias spark_catalog.default.deltaTable
+- Relation spark_catalog.default.deltatable[col1#422,col2#423,col3#424,col4#425] parquet
```
- Non-delta table:
```
CREATE TABLE parquetTable (col1 INT, col2 INT, col3 INT, col4 INT) USING parquet;
SELECT col1, col2, col3, NULLIF('','') AS col4
FROM parquetTable
UNION ALL
SELECT col2, col2, null AS col3, col4
FROM parquetTable;
```
In this case, we `ResolveReferences` (deduplication of `Union` children outputs) before `WidenSetOperationTypes` and thus plan looks like:
```
Union false, false
:- Project [col1#2, col2#3, col3#4, cast(col4#0 as bigint) AS col4#11L]
: +- Project [col1#2, col2#3, col3#4, nullif(, ) AS col4#0]
: +- SubqueryAlias spark_catalog.default.parquettable
: +- Relation spark_catalog.default.parquettable[col1#2,col2#3,col3#4,col4#5] parquet
+- Project [col2#7, col2#10, cast(col3#1 as int) AS col3#12, cast(col4#9 as bigint) AS col4#13L]
+- Project [col2#7, col2#7 AS col2#10, col3#1, col4#9]
+- Project [col2#7, col2#7, null AS col3#1, col4#9]
+- SubqueryAlias spark_catalog.default.parquettable
+- Relation spark_catalog.default.parquettable[col1#6,col2#7,col3#8,col4#9] parquet
```
In this issue I propose that we align those two by enforcing type coercion to happen before deduplication.
### Why are the changes needed?
To make `UNION` with different underlying table providers producing consistent plans.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests + existing ones.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #51172 from mihailoale-db/uniondeduplicationfix.
Authored-by: mihailoale-db <mi...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../spark/sql/catalyst/analysis/Analyzer.scala | 4 ++-
.../plans/logical/basicLogicalOperators.scala | 23 ++++++++++------
.../org/apache/spark/sql/internal/SQLConf.scala | 11 ++++++++
.../sql-tests/analyzer-results/union.sql.out | 32 ++++++++++++++++++++++
.../src/test/resources/sql-tests/inputs/union.sql | 9 ++++++
.../test/resources/sql-tests/results/union.sql.out | 28 +++++++++++++++++++
6 files changed, 97 insertions(+), 10 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index bcedceee6e70..f9272cc03735 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -1573,7 +1573,9 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
case u @ Union(children, _, _)
// if there are duplicate output columns, give them unique expr ids
- if children.exists(c => c.output.map(_.exprId).distinct.length < c.output.length) =>
+ if (u.allChildrenCompatible &&
+ conf.getConf(SQLConf.ENFORCE_TYPE_COERCION_BEFORE_UNION_DEDUPLICATION)) &&
+ children.exists(c => c.output.map(_.exprId).distinct.length < c.output.length) =>
val newChildren = children.map { c =>
if (c.output.map(_.exprId).distinct.length < c.output.length) {
val existingExprIds = mutable.HashSet[ExprId]()
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index 03d1a9fca3cc..ba56e8599e56 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -603,18 +603,23 @@ case class Union(
}
override lazy val resolved: Boolean = {
- // allChildrenCompatible needs to be evaluated after childrenResolved
- def allChildrenCompatible: Boolean =
- children.tail.forall( child =>
- // compare the attribute number with the first child
- child.output.length == children.head.output.length &&
- // compare the data types with the first child
- child.output.zip(children.head.output).forall {
- case (l, r) => DataType.equalsStructurally(l.dataType, r.dataType, true)
- })
children.length > 1 && !(byName || allowMissingCol) && childrenResolved && allChildrenCompatible
}
+ /**
+ * Checks whether the child outputs are compatible by using `DataType.equalsStructurally`. Do
+ * that by comparing the size of the output with the size of the first child's output and by
+ * comparing output data types with the data types of the first child's output.
+ *
+ * This method needs to be evaluated after `childrenResolved`.
+ */
+ def allChildrenCompatible: Boolean = childrenResolved && children.tail.forall { child =>
+ child.output.length == children.head.output.length &&
+ child.output.zip(children.head.output).forall {
+ case (l, r) => DataType.equalsStructurally(l.dataType, r.dataType, true)
+ }
+ }
+
override protected def withNewChildrenInternal(newChildren: IndexedSeq[LogicalPlan]): Union =
copy(children = newChildren)
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index e873a1897445..0fe08fc719a8 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -5960,6 +5960,17 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val ENFORCE_TYPE_COERCION_BEFORE_UNION_DEDUPLICATION =
+ buildConf("spark.sql.enforceTypeCoercionBeforeUnionDeduplication.enabled")
+ .internal()
+ .doc(
+ "When set to true, we enforce type coercion to run before deduplication of UNION " +
+ "children outputs. Otherwise, order is relative to rule ordering."
+ )
+ .version("4.1.0")
+ .booleanConf
+ .createWithDefault(true)
+
/**
* Holds information about keys that have been deprecated.
*
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/union.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/union.sql.out
index dfba3688f0b7..a2dfb6e47a57 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/union.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/union.sql.out
@@ -15,6 +15,12 @@ CreateViewCommand `t2`, VALUES (1.0, 1), (2.0, 4) tbl(c1, c2), false, true, Loca
+- LocalRelation [c1#x, c2#x]
+-- !query
+CREATE TABLE jsonTable (col1 INT, col2 INT, col3 INT, col4 INT) USING json
+-- !query analysis
+CreateDataSourceTableCommand `spark_catalog`.`default`.`jsonTable`, false
+
+
-- !query
SELECT *
FROM (SELECT * FROM t1
@@ -241,6 +247,25 @@ Aggregate [sum(v#x) AS sum(v)#x]
+- LocalRelation [v#x]
+-- !query
+SELECT col1, col2, col3, NULLIF('','') AS col4
+FROM jsonTable
+UNION ALL
+SELECT col2, col2, null AS col3, col4
+FROM jsonTable
+-- !query analysis
+Union false, false
+:- Project [col1#x, col2#x, col3#x, cast(col4#x as bigint) AS col4#xL]
+: +- Project [col1#x, col2#x, col3#x, nullif(, ) AS col4#x]
+: +- SubqueryAlias spark_catalog.default.jsontable
+: +- Relation spark_catalog.default.jsontable[col1#x,col2#x,col3#x,col4#x] json
++- Project [col2#x, col2#x AS col2#x, col3#x, col4#xL]
+ +- Project [col2#x, col2#x, cast(col3#x as int) AS col3#x, cast(col4#x as bigint) AS col4#xL]
+ +- Project [col2#x, col2#x, null AS col3#x, col4#x]
+ +- SubqueryAlias spark_catalog.default.jsontable
+ +- Relation spark_catalog.default.jsontable[col1#x,col2#x,col3#x,col4#x] json
+
+
-- !query
DROP VIEW IF EXISTS t1
-- !query analysis
@@ -275,3 +300,10 @@ DropTempViewCommand p2
DROP VIEW IF EXISTS p3
-- !query analysis
DropTempViewCommand p3
+
+
+-- !query
+DROP TABLE IF EXISTS jsonTable
+-- !query analysis
+DropTable true, false
++- ResolvedIdentifier V2SessionCatalog(spark_catalog), default.jsonTable
diff --git a/sql/core/src/test/resources/sql-tests/inputs/union.sql b/sql/core/src/test/resources/sql-tests/inputs/union.sql
index ab81cc7bbabb..01b53c613274 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/union.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/union.sql
@@ -1,5 +1,6 @@
CREATE OR REPLACE TEMPORARY VIEW t1 AS VALUES (1, 'a'), (2, 'b') tbl(c1, c2);
CREATE OR REPLACE TEMPORARY VIEW t2 AS VALUES (1.0, 1), (2.0, 4) tbl(c1, c2);
+CREATE TABLE jsonTable (col1 INT, col2 INT, col3 INT, col4 INT) USING json;
-- Simple Union
SELECT *
@@ -59,6 +60,13 @@ SELECT SUM(t.v) FROM (
SELECT v + v AS v FROM t3
) t;
+-- SPARK-52462: UNION should produce consistent results with different underlying table providers.
+SELECT col1, col2, col3, NULLIF('','') AS col4
+FROM jsonTable
+UNION ALL
+SELECT col2, col2, null AS col3, col4
+FROM jsonTable;
+
-- Clean-up
DROP VIEW IF EXISTS t1;
DROP VIEW IF EXISTS t2;
@@ -66,3 +74,4 @@ DROP VIEW IF EXISTS t3;
DROP VIEW IF EXISTS p1;
DROP VIEW IF EXISTS p2;
DROP VIEW IF EXISTS p3;
+DROP TABLE IF EXISTS jsonTable;
diff --git a/sql/core/src/test/resources/sql-tests/results/union.sql.out b/sql/core/src/test/resources/sql-tests/results/union.sql.out
index ce4d3421ae83..2086a81c9e29 100644
--- a/sql/core/src/test/resources/sql-tests/results/union.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/union.sql.out
@@ -15,6 +15,14 @@ struct<>
+-- !query
+CREATE TABLE jsonTable (col1 INT, col2 INT, col3 INT, col4 INT) USING json
+-- !query schema
+struct<>
+-- !query output
+
+
+
-- !query
SELECT *
FROM (SELECT * FROM t1
@@ -200,6 +208,18 @@ struct<sum(v):decimal(21,0)>
3
+-- !query
+SELECT col1, col2, col3, NULLIF('','') AS col4
+FROM jsonTable
+UNION ALL
+SELECT col2, col2, null AS col3, col4
+FROM jsonTable
+-- !query schema
+struct<col1:int,col2:int,col3:int,col4:bigint>
+-- !query output
+
+
+
-- !query
DROP VIEW IF EXISTS t1
-- !query schema
@@ -246,3 +266,11 @@ DROP VIEW IF EXISTS p3
struct<>
-- !query output
+
+
+-- !query
+DROP TABLE IF EXISTS jsonTable
+-- !query schema
+struct<>
+-- !query output
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org