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 2026/10/14 14:56:18 UTC
(spark) branch master updated: [SPARK-60038][SQL] Merge nested nullability of all values in UNPIVOT value columns
This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 4ccf07b6d91e [SPARK-60038][SQL] Merge nested nullability of all values in UNPIVOT value columns
4ccf07b6d91e is described below
commit 4ccf07b6d91e2b126b7001741778ee28f22c6609
Author: Mihailo Aleksic <mi...@databricks.com>
AuthorDate: Wed Oct 14 14:56:01 2026 +0000
[SPARK-60038][SQL] Merge nested nullability of all values in UNPIVOT value columns
### What changes were proposed in this pull request?
`Unpivot.valuesTypeCoercioned` compares the value types with `DataTypeUtils.sameType`, which ignores nullability. When the values at a position differ only in nested nullability (array `containsNull`, map `valueContainsNull`, struct field `nullable`), no coercion
is needed, and `UnpivotTransformer` typed the value column with the first value's data type (`values.head(index).dataType`). Only top-level nullability was merged across the values, so a `NULL` nested value of a later value could be returned as a non-`NULL` value.
This PR makes `UnpivotTransformer` merge the nested nullability of all values into the value column type with `StructType.unionLikeMerge`, the same way `Union` merges the output of its children. The values are not cast, and the value column keeps the nested field
names and metadata of the first value:
```
Expand [[store#x, h1, h1#x], [store#x, h2, h2#x]], [store#x, half#x, b#x]
```
It also:
- Requires in `Unpivot.canBeCoercioned` that every value group has one value per value column, so a malformed multi-value `UNPIVOT` reports `UNPIVOT_VALUE_SIZE_MISMATCH` instead of failing with an `IndexOutOfBoundsException` during coercion.
- Makes the single-pass `UnpivotResolver` validate the value sizes first and apply `UnpivotTypeCoercion` only when `valuesTypeCoercioned` is false, like the fixed-point `UnpivotCoercion` rule.
- Adds `spark.sql.legacy.unpivot.useFirstValueNestedNullability` to restore the previous behavior.
### Why are the changes needed?
To fix the correctness issue.
### Does this PR introduce _any_ user-facing change?
Yes, it changes the end result.
### How was this patch tested?
Added tests.
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Closes #59271 from mihailoale-db/unpivot-nested-nullability.
Authored-by: Mihailo Aleksic <mi...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
docs/sql-migration-guide.md | 1 +
.../sql/catalyst/analysis/UnpivotTransformer.scala | 22 +++-
.../analysis/resolver/UnpivotResolver.scala | 15 ++-
.../plans/logical/basicLogicalOperators.scala | 6 +-
.../org/apache/spark/sql/internal/SQLConf.scala | 13 +++
.../sql-tests/analyzer-results/unpivot.sql.out | 126 +++++++++++++++++++++
.../test/resources/sql-tests/inputs/unpivot.sql | 47 ++++++++
.../resources/sql-tests/results/unpivot.sql.out | 95 ++++++++++++++++
.../org/apache/spark/sql/DatasetUnpivotSuite.scala | 115 ++++++++++++++++++-
9 files changed, 430 insertions(+), 10 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index 31c90b751fe5..e80f35d82c82 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -47,6 +47,7 @@ license: |
- Since Spark 4.4, `spark.sql.sources.v2.bucketing.partition.filter.enabled` defaults to `true`. In a storage-partitioned join, the partition key groups pushed down to both sides may now be narrowed to those that can produce output for the join type, instead of always taking the union of the two sides' groups: an inner or semi join may keep only the groups present on both sides, a join that keeps or tests every left row (left outer, left anti, left single, existence) keeps the left side' [...]
- Since Spark 4.4, `spark.sql.sources.v2.bucketing.partitionKeyOrdering.enabled` defaults to `true`. A V2 scan that reports a keyed partitioning but no explicit ordering now also reports itself sorted by its partition key expressions, because every row of such a partition evaluates them to the same value. Partition transforms such as `days(ts)` or `bucket(8, id)` are left out of that ordering. This lets Spark drop a `Sort` it would otherwise place above the scan. An aggregate that groups [...]
- Since Spark 4.4, `spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled` defaults to `true`. When `GroupPartitionsExec` merges several input partitions that share one partition key value into a single output partition, it now keeps sort orders over the partition key expressions in the ordering it reports, since those expressions are constant within the merged partition. Sort orders over partition transforms such as `days(ts)` or `bucket(8, id)` are left out. Orders over [...]
+- Since Spark 4.4, the value columns of `UNPIVOT` and `Dataset.unpivot` merge the nested nullability (array element, map value, and struct field nullability) of all unpivoted values. Previously, a value column took the nested nullability of the first listed value, so a `NULL` array element, map value, or struct field of a later value could be returned as a non-`NULL` value such as `0`, and `IS NULL` on it returned `false`. A view created by an earlier Spark version over such an `UNPIVOT` [...]
## Upgrading from Spark SQL 4.2 to 4.3
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnpivotTransformer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnpivotTransformer.scala
index 2bb6901148b9..458ae959b69b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnpivotTransformer.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/UnpivotTransformer.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.catalyst.analysis
+import org.apache.spark.sql.catalyst.SQLConfHelper
import org.apache.spark.sql.catalyst.expressions.{
AttributeReference,
Expression,
@@ -24,12 +25,13 @@ import org.apache.spark.sql.catalyst.expressions.{
NamedExpression
}
import org.apache.spark.sql.catalyst.plans.logical.{Expand, LogicalPlan}
-import org.apache.spark.sql.types.StringType
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{DataType, StringType, StructType}
/**
* Object used to transform the given [[Unpivot]] node to an [[Expand]] node.
*/
-object UnpivotTransformer {
+object UnpivotTransformer extends SQLConfHelper {
/**
* Construct an [[Expand]] node from the given [[Unpivot]] node. Do that by:
@@ -64,7 +66,7 @@ object UnpivotTransformer {
case (valueColumnName, index) =>
AttributeReference(
valueColumnName,
- values.head(index).dataType,
+ valueDataType(values.map(_(index).dataType)),
values.map(_(index)).exists(_.nullable)
)()
}
@@ -73,4 +75,18 @@ object UnpivotTransformer {
Expand(expressions, output, child)
}
+
+ /**
+ * Get the data type of a value column out of the data types of its values, which are the same
+ * up to nullability. Like [[Union]], merge the nested nullability of all values without casting
+ * them, keeping the field names and metadata of the first value. In the legacy mode, take the
+ * data type of the first value.
+ */
+ private def valueDataType(dataTypes: Seq[DataType]): DataType = {
+ if (conf.getConf(SQLConf.LEGACY_UNPIVOT_USE_FIRST_VALUE_NESTED_NULLABILITY)) {
+ dataTypes.head
+ } else {
+ dataTypes.reduce(StructType.unionLikeMerge)
+ }
+ }
}
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnpivotResolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnpivotResolver.scala
index 62298e74d0cc..68660be2059a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnpivotResolver.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/UnpivotResolver.scala
@@ -43,8 +43,9 @@ class UnpivotResolver(operatorResolver: Resolver, expressionResolver: Expression
* 1. Resolve the child operator.
* 2. Resolve the ids and values expressions based on their presence (see respective scala docs
* for more info).
- * 3. Perform type coercion on the values expressions.
- * 4. Validate the lengths of the values expressions.
+ * 3. Validate the lengths of the values expressions.
+ * 4. Perform type coercion on the values expressions, unless their types are already the same
+ * up to nullability.
* 5. Use the [[UnpivotTransformer]] to create the resolved operator (see its scala doc for more
* info).
* 6. Map the output attributes to new ones with unique expression ids. This is needed since the
@@ -91,11 +92,15 @@ class UnpivotResolver(operatorResolver: Resolver, expressionResolver: Expression
val partiallyResolvedUnpivot =
unpivot.copy(ids = Some(resolvedIds), values = Some(resolvedValues))
- val typeCoercedUnpivot = UnpivotTypeCoercion(partiallyResolvedUnpivot)
+ validateValuesLength(partiallyResolvedUnpivot.values.get, partiallyResolvedUnpivot)
- validateValuesTypeCoercioned(typeCoercedUnpivot)
+ val typeCoercedUnpivot = if (partiallyResolvedUnpivot.valuesTypeCoercioned) {
+ partiallyResolvedUnpivot
+ } else {
+ UnpivotTypeCoercion(partiallyResolvedUnpivot)
+ }
- validateValuesLength(typeCoercedUnpivot.values.get, typeCoercedUnpivot)
+ validateValuesTypeCoercioned(typeCoercedUnpivot)
val resolvedOperator: Expand = UnpivotTransformer(
ids = typeCoercedUnpivot.ids.get,
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 01a1c3375594..85edd150b8ed 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
@@ -1759,12 +1759,16 @@ case class Unpivot(
def canBeCoercioned: Boolean = values.exists(_.nonEmpty) &&
values.exists(_.forall(_.forall(_.resolved))) &&
+ // values are coerced by position, so every value group must have one value per value column,
+ // otherwise CheckAnalysis reports UNPIVOT_VALUE_SIZE_MISMATCH
+ values.exists(_.forall(_.length == valueColumnNames.length)) &&
// when no ids are given, values must be Attributes (column names) to allow detecting ids
// coercion will add aliases, would disallow detecting ids, so defer coercion after id detection
ids.exists(_.forall(_.resolved))
def valuesTypeCoercioned: Boolean = canBeCoercioned &&
- // all inner values at position idx must have the same data type
+ // all inner values at position idx must have the same data type, ignoring nullability;
+ // UnpivotTransformer merges the nullability of the values into the value column
values.get.head.zipWithIndex.forall { case (v, idx) =>
values.get.tail.forall(vals => DataTypeUtils.sameType(vals(idx).dataType, v.dataType))
}
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 2a89d3de43ba..dca0ca260da6 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
@@ -7713,6 +7713,19 @@ object SQLConf {
.booleanConf
.createWithDefault(false)
+ val LEGACY_UNPIVOT_USE_FIRST_VALUE_NESTED_NULLABILITY =
+ buildConf("spark.sql.legacy.unpivot.useFirstValueNestedNullability")
+ .internal()
+ .doc("When true, the value columns of UNPIVOT and Dataset.unpivot take the nested " +
+ "nullability (array element, map value, and struct field nullability) of the first " +
+ "listed value, which is the behavior in Spark 4.3 and earlier. A NULL nested value of " +
+ "a later value can then be returned as a non-NULL value. When false, the nested " +
+ "nullability of all values is merged.")
+ .version("4.4.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
+
val LEGACY_CREATE_HIVE_TABLE_BY_DEFAULT =
buildConf("spark.sql.legacy.createHiveTableByDefault")
.internal()
diff --git a/sql/core/src/test/resources/sql-tests/analyzer-results/unpivot.sql.out b/sql/core/src/test/resources/sql-tests/analyzer-results/unpivot.sql.out
index 230cffec9a42..d82425176d1f 100644
--- a/sql/core/src/test/resources/sql-tests/analyzer-results/unpivot.sql.out
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/unpivot.sql.out
@@ -186,3 +186,129 @@ Sort [kind#x ASC NULLS FIRST], true
+- Project [cast(id#x as int) AS id#x, cast(date1#x as date) AS date1#x, cast(int1#x as int) AS int1#x, cast(date2#x as date) AS date2#x, cast(int2#x as int) AS int2#x, cast(date3#x as date) AS date3#x, cast(int3#x as int) AS int3#x]
+- Project [1 AS id#x, cast(2026-01-01 as date) AS date1#x, cast(null as int) AS int1#x, cast(null as date) AS date2#x, 2 AS int2#x, cast(null as date) AS date3#x, cast(null as int) AS int3#x]
+- OneRowRelation
+
+
+-- !query
+create temporary view monthlySales as select * from values
+ ("north", 10, 20, 30, cast(null as int)),
+ ("south", 5, 6, NULL, NULL)
+ as monthlySales(store, jan, feb, mar, apr)
+-- !query analysis
+CreateViewCommand `monthlySales`, select * from values
+ ("north", 10, 20, 30, cast(null as int)),
+ ("south", 5, 6, NULL, NULL)
+ as monthlySales(store, jan, feb, mar, apr), false, false, LocalTempView, UNSUPPORTED, true
+ +- Project [store#x, jan#x, feb#x, mar#x, apr#x]
+ +- SubqueryAlias monthlySales
+ +- LocalRelation [store#x, jan#x, feb#x, mar#x, apr#x]
+
+
+-- !query
+SELECT store, half, b.lo, b.hi, b.hi IS NULL AS hi_is_null
+FROM (
+ SELECT store, named_struct('lo', jan, 'hi', feb) AS h1, named_struct('lo', mar, 'hi', apr) AS h2
+ FROM monthlySales
+)
+UNPIVOT (b FOR half IN (h1, h2))
+ORDER BY store, half
+-- !query analysis
+Sort [store#x ASC NULLS FIRST, half#x ASC NULLS FIRST], true
++- Project [store#x, half#x, b#x.lo AS lo#x, b#x.hi AS hi#x, isnull(b#x.hi) AS hi_is_null#x]
+ +- Filter isnotnull(coalesce(b#x))
+ +- Expand [[store#x, h1, h1#x], [store#x, h2, h2#x]], [store#x, half#x, b#x]
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [store#x, named_struct(lo, jan#x, hi, feb#x) AS h1#x, named_struct(lo, mar#x, hi, apr#x) AS h2#x]
+ +- SubqueryAlias monthlysales
+ +- View (`monthlySales`, [store#x, jan#x, feb#x, mar#x, apr#x])
+ +- Project [cast(store#x as string) AS store#x, cast(jan#x as int) AS jan#x, cast(feb#x as int) AS feb#x, cast(mar#x as int) AS mar#x, cast(apr#x as int) AS apr#x]
+ +- Project [store#x, jan#x, feb#x, mar#x, apr#x]
+ +- SubqueryAlias monthlySales
+ +- LocalRelation [store#x, jan#x, feb#x, mar#x, apr#x]
+
+
+-- !query
+SELECT store, half, b.lo, b.hi, b.hi IS NULL AS hi_is_null
+FROM (
+ SELECT store, named_struct('lo', jan, 'hi', feb) AS h1, named_struct('lo', mar, 'hi', apr) AS h2
+ FROM monthlySales
+)
+UNPIVOT (b FOR half IN (h2, h1))
+ORDER BY store, half
+-- !query analysis
+Sort [store#x ASC NULLS FIRST, half#x ASC NULLS FIRST], true
++- Project [store#x, half#x, b#x.lo AS lo#x, b#x.hi AS hi#x, isnull(b#x.hi) AS hi_is_null#x]
+ +- Filter isnotnull(coalesce(b#x))
+ +- Expand [[store#x, h2, h2#x], [store#x, h1, h1#x]], [store#x, half#x, b#x]
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [store#x, named_struct(lo, jan#x, hi, feb#x) AS h1#x, named_struct(lo, mar#x, hi, apr#x) AS h2#x]
+ +- SubqueryAlias monthlysales
+ +- View (`monthlySales`, [store#x, jan#x, feb#x, mar#x, apr#x])
+ +- Project [cast(store#x as string) AS store#x, cast(jan#x as int) AS jan#x, cast(feb#x as int) AS feb#x, cast(mar#x as int) AS mar#x, cast(apr#x as int) AS apr#x]
+ +- Project [store#x, jan#x, feb#x, mar#x, apr#x]
+ +- SubqueryAlias monthlySales
+ +- LocalRelation [store#x, jan#x, feb#x, mar#x, apr#x]
+
+
+-- !query
+SELECT store, half, m, m IS NULL AS m_is_null
+FROM (SELECT store, array(jan, feb) AS h1, array(mar, apr) AS h2 FROM monthlySales)
+UNPIVOT (months FOR half IN (h1, h2))
+LATERAL VIEW explode(months) e AS m
+ORDER BY store, half, m
+-- !query analysis
+Sort [store#x ASC NULLS FIRST, half#x ASC NULLS FIRST, m#x ASC NULLS FIRST], true
++- Project [store#x, half#x, m#x, isnull(m#x) AS m_is_null#x]
+ +- Generate explode(months#x), false, e, [m#x]
+ +- Filter isnotnull(coalesce(months#x))
+ +- Expand [[store#x, h1, h1#x], [store#x, h2, h2#x]], [store#x, half#x, months#x]
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [store#x, array(jan#x, feb#x) AS h1#x, array(mar#x, apr#x) AS h2#x]
+ +- SubqueryAlias monthlysales
+ +- View (`monthlySales`, [store#x, jan#x, feb#x, mar#x, apr#x])
+ +- Project [cast(store#x as string) AS store#x, cast(jan#x as int) AS jan#x, cast(feb#x as int) AS feb#x, cast(mar#x as int) AS mar#x, cast(apr#x as int) AS apr#x]
+ +- Project [store#x, jan#x, feb#x, mar#x, apr#x]
+ +- SubqueryAlias monthlySales
+ +- LocalRelation [store#x, jan#x, feb#x, mar#x, apr#x]
+
+
+-- !query
+SELECT store, half, m['k'] AS v, m['k'] IS NULL AS v_is_null
+FROM (SELECT store, map('k', jan) AS m1, map('k', mar) AS m2 FROM monthlySales)
+UNPIVOT (m FOR half IN (m1, m2))
+ORDER BY store, half
+-- !query analysis
+Sort [store#x ASC NULLS FIRST, half#x ASC NULLS FIRST], true
++- Project [store#x, half#x, m#x[k] AS v#x, isnull(m#x[k]) AS v_is_null#x]
+ +- Filter isnotnull(coalesce(m#x))
+ +- Expand [[store#x, m1, m1#x], [store#x, m2, m2#x]], [store#x, half#x, m#x]
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [store#x, map(k, jan#x) AS m1#x, map(k, mar#x) AS m2#x]
+ +- SubqueryAlias monthlysales
+ +- View (`monthlySales`, [store#x, jan#x, feb#x, mar#x, apr#x])
+ +- Project [cast(store#x as string) AS store#x, cast(jan#x as int) AS jan#x, cast(feb#x as int) AS feb#x, cast(mar#x as int) AS mar#x, cast(apr#x as int) AS apr#x]
+ +- Project [store#x, jan#x, feb#x, mar#x, apr#x]
+ +- SubqueryAlias monthlySales
+ +- LocalRelation [store#x, jan#x, feb#x, mar#x, apr#x]
+
+
+-- !query
+SELECT store, half, s.hi, s.hi IS NULL AS hi_is_null, a
+FROM (
+ SELECT store, named_struct('hi', feb) AS s1, jan AS a1, named_struct('hi', apr) AS s2, mar AS a2
+ FROM monthlySales
+)
+UNPIVOT ((s, a) FOR half IN ((s1, a1) AS x, (s2, a2) AS y))
+ORDER BY store, half
+-- !query analysis
+Sort [store#x ASC NULLS FIRST, half#x ASC NULLS FIRST], true
++- Project [store#x, half#x, s#x.hi AS hi#x, isnull(s#x.hi) AS hi_is_null#x, a#x]
+ +- Filter (isnotnull(s#x) OR isnotnull(a#x))
+ +- Expand [[store#x, x, s1#x, a1#x], [store#x, y, s2#x, a2#x]], [store#x, half#x, s#x, a#x]
+ +- SubqueryAlias __auto_generated_subquery_name
+ +- Project [store#x, named_struct(hi, feb#x) AS s1#x, jan#x AS a1#x, named_struct(hi, apr#x) AS s2#x, mar#x AS a2#x]
+ +- SubqueryAlias monthlysales
+ +- View (`monthlySales`, [store#x, jan#x, feb#x, mar#x, apr#x])
+ +- Project [cast(store#x as string) AS store#x, cast(jan#x as int) AS jan#x, cast(feb#x as int) AS feb#x, cast(mar#x as int) AS mar#x, cast(apr#x as int) AS apr#x]
+ +- Project [store#x, jan#x, feb#x, mar#x, apr#x]
+ +- SubqueryAlias monthlySales
+ +- LocalRelation [store#x, jan#x, feb#x, mar#x, apr#x]
diff --git a/sql/core/src/test/resources/sql-tests/inputs/unpivot.sql b/sql/core/src/test/resources/sql-tests/inputs/unpivot.sql
index 43092080ec9f..e0d26eb1762a 100644
--- a/sql/core/src/test/resources/sql-tests/inputs/unpivot.sql
+++ b/sql/core/src/test/resources/sql-tests/inputs/unpivot.sql
@@ -71,3 +71,50 @@ UNPIVOT INCLUDE NULLS (
)
)
ORDER BY kind;
+
+
+-- jan and feb are NOT NULL, mar and apr are nullable
+create temporary view monthlySales as select * from values
+ ("north", 10, 20, 30, cast(null as int)),
+ ("south", 5, 6, NULL, NULL)
+ as monthlySales(store, jan, feb, mar, apr);
+
+-- struct values: the value column merges the field nullability of all values
+SELECT store, half, b.lo, b.hi, b.hi IS NULL AS hi_is_null
+FROM (
+ SELECT store, named_struct('lo', jan, 'hi', feb) AS h1, named_struct('lo', mar, 'hi', apr) AS h2
+ FROM monthlySales
+)
+UNPIVOT (b FOR half IN (h1, h2))
+ORDER BY store, half;
+
+-- the result does not depend on the order of the values
+SELECT store, half, b.lo, b.hi, b.hi IS NULL AS hi_is_null
+FROM (
+ SELECT store, named_struct('lo', jan, 'hi', feb) AS h1, named_struct('lo', mar, 'hi', apr) AS h2
+ FROM monthlySales
+)
+UNPIVOT (b FOR half IN (h2, h1))
+ORDER BY store, half;
+
+-- array values: the value column merges the element nullability of all values
+SELECT store, half, m, m IS NULL AS m_is_null
+FROM (SELECT store, array(jan, feb) AS h1, array(mar, apr) AS h2 FROM monthlySales)
+UNPIVOT (months FOR half IN (h1, h2))
+LATERAL VIEW explode(months) e AS m
+ORDER BY store, half, m;
+
+-- map values: the value column merges the value nullability of all values
+SELECT store, half, m['k'] AS v, m['k'] IS NULL AS v_is_null
+FROM (SELECT store, map('k', jan) AS m1, map('k', mar) AS m2 FROM monthlySales)
+UNPIVOT (m FOR half IN (m1, m2))
+ORDER BY store, half;
+
+-- multiple value columns: each value column merges the nullability of its own position
+SELECT store, half, s.hi, s.hi IS NULL AS hi_is_null, a
+FROM (
+ SELECT store, named_struct('hi', feb) AS s1, jan AS a1, named_struct('hi', apr) AS s2, mar AS a2
+ FROM monthlySales
+)
+UNPIVOT ((s, a) FOR half IN ((s1, a1) AS x, (s2, a2) AS y))
+ORDER BY store, half;
diff --git a/sql/core/src/test/resources/sql-tests/results/unpivot.sql.out b/sql/core/src/test/resources/sql-tests/results/unpivot.sql.out
index 185f747b4d4e..af3cf853c611 100644
--- a/sql/core/src/test/resources/sql-tests/results/unpivot.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/unpivot.sql.out
@@ -158,3 +158,98 @@ struct<id:int,kind:string,date_value:date,int_value:int>
1 all_null NULL NULL
1 date_only 2026-01-01 NULL
1 int_only NULL 2
+
+
+-- !query
+create temporary view monthlySales as select * from values
+ ("north", 10, 20, 30, cast(null as int)),
+ ("south", 5, 6, NULL, NULL)
+ as monthlySales(store, jan, feb, mar, apr)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT store, half, b.lo, b.hi, b.hi IS NULL AS hi_is_null
+FROM (
+ SELECT store, named_struct('lo', jan, 'hi', feb) AS h1, named_struct('lo', mar, 'hi', apr) AS h2
+ FROM monthlySales
+)
+UNPIVOT (b FOR half IN (h1, h2))
+ORDER BY store, half
+-- !query schema
+struct<store:string,half:string,lo:int,hi:int,hi_is_null:boolean>
+-- !query output
+north h1 10 20 false
+north h2 30 NULL true
+south h1 5 6 false
+south h2 NULL NULL true
+
+
+-- !query
+SELECT store, half, b.lo, b.hi, b.hi IS NULL AS hi_is_null
+FROM (
+ SELECT store, named_struct('lo', jan, 'hi', feb) AS h1, named_struct('lo', mar, 'hi', apr) AS h2
+ FROM monthlySales
+)
+UNPIVOT (b FOR half IN (h2, h1))
+ORDER BY store, half
+-- !query schema
+struct<store:string,half:string,lo:int,hi:int,hi_is_null:boolean>
+-- !query output
+north h1 10 20 false
+north h2 30 NULL true
+south h1 5 6 false
+south h2 NULL NULL true
+
+
+-- !query
+SELECT store, half, m, m IS NULL AS m_is_null
+FROM (SELECT store, array(jan, feb) AS h1, array(mar, apr) AS h2 FROM monthlySales)
+UNPIVOT (months FOR half IN (h1, h2))
+LATERAL VIEW explode(months) e AS m
+ORDER BY store, half, m
+-- !query schema
+struct<store:string,half:string,m:int,m_is_null:boolean>
+-- !query output
+north h1 10 false
+north h1 20 false
+north h2 NULL true
+north h2 30 false
+south h1 5 false
+south h1 6 false
+south h2 NULL true
+south h2 NULL true
+
+
+-- !query
+SELECT store, half, m['k'] AS v, m['k'] IS NULL AS v_is_null
+FROM (SELECT store, map('k', jan) AS m1, map('k', mar) AS m2 FROM monthlySales)
+UNPIVOT (m FOR half IN (m1, m2))
+ORDER BY store, half
+-- !query schema
+struct<store:string,half:string,v:int,v_is_null:boolean>
+-- !query output
+north m1 10 false
+north m2 30 false
+south m1 5 false
+south m2 NULL true
+
+
+-- !query
+SELECT store, half, s.hi, s.hi IS NULL AS hi_is_null, a
+FROM (
+ SELECT store, named_struct('hi', feb) AS s1, jan AS a1, named_struct('hi', apr) AS s2, mar AS a2
+ FROM monthlySales
+)
+UNPIVOT ((s, a) FOR half IN ((s1, a1) AS x, (s2, a2) AS y))
+ORDER BY store, half
+-- !query schema
+struct<store:string,half:string,hi:int,hi_is_null:boolean,a:int>
+-- !query output
+north x 20 false 10
+north y NULL true 30
+south x 6 false 5
+south y NULL true NULL
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala
index 97ab31c3405a..484e07d7b523 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala
@@ -17,7 +17,8 @@
package org.apache.spark.sql
-import org.apache.spark.sql.functions.{length, struct, sum}
+import org.apache.spark.sql.catalyst.expressions.Cast
+import org.apache.spark.sql.functions.{array, explode, length, lit, map, struct, sum}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
@@ -667,6 +668,118 @@ class DatasetUnpivotSuite extends SharedSparkSession {
}
}
+ test("unpivot sql with value groups of different sizes") {
+ // b differs from a only in nested nullability, d differs from a in element type
+ Seq("(a), (b, c)", "(b, c), (a)", "(a), (d, c)", "(d, c), (a)").foreach { groups =>
+ withClue(groups) {
+ val e = intercept[AnalysisException] {
+ spark.sql(s"""SELECT * FROM (
+ | SELECT array(1) a, array(cast(null as int)) b, 1 c, array(1L) d
+ |)
+ |UNPIVOT (
+ | (val1, val2) FOR col IN ($groups)
+ |);
+ |""".stripMargin)
+ }
+ checkError(
+ exception = e,
+ condition = "UNPIVOT_VALUE_SIZE_MISMATCH",
+ parameters = Map("names" -> "2"))
+ }
+ }
+ }
+
+ test("unpivot merges nested nullability of all values") {
+ // jan and feb are NOT NULL, mar and apr are nullable
+ val df = Seq(
+ ("north", 10, 20, Some(30), Option.empty[Int]),
+ ("south", 5, 6, Option.empty[Int], Option.empty[Int])
+ ).toDF("store", "jan", "feb", "mar", "apr")
+ val nullableLoHi = new StructType().add("lo", IntegerType).add("hi", IntegerType)
+
+ Seq(
+ Array[Column]($"h1", $"h2") -> Seq(
+ Row("north", "h1", 10, 20, false), Row("north", "h2", 30, null, true),
+ Row("south", "h1", 5, 6, false), Row("south", "h2", null, null, true)),
+ Array[Column]($"h2", $"h1") -> Seq(
+ Row("north", "h2", 30, null, true), Row("north", "h1", 10, 20, false),
+ Row("south", "h2", null, null, true), Row("south", "h1", 5, 6, false))
+ ).foreach { case (values, expected) =>
+ val unpivoted = df
+ .select(
+ $"store",
+ struct($"jan".as("lo"), $"feb".as("hi")).as("h1"),
+ struct($"mar".as("lo"), $"apr".as("hi")).as("h2"))
+ .unpivot(Array($"store"), values, "half", "b")
+ assert(unpivoted.schema("b").dataType === nullableLoHi)
+ checkAnswer(
+ unpivoted.select($"store", $"half", $"b.lo", $"b.hi", $"b.hi".isNull),
+ expected)
+ }
+
+ val arrays = df
+ .select($"store", array($"jan", $"feb").as("h1"), array($"mar", $"apr").as("h2"))
+ .unpivot(Array($"store"), Array($"h1", $"h2"), "half", "months")
+ assert(arrays.schema("months").dataType === ArrayType(IntegerType, containsNull = true))
+ checkAnswer(
+ arrays.select($"store", $"half", explode($"months").as("m")).select($"half", $"m".isNull),
+ Seq(
+ Row("h1", false), Row("h1", false), Row("h2", false), Row("h2", true),
+ Row("h1", false), Row("h1", false), Row("h2", true), Row("h2", true)))
+
+ val maps = df
+ .select($"store", map(lit("k"), $"jan").as("m1"), map(lit("k"), $"mar").as("m2"))
+ .unpivot(Array($"store"), Array($"m1", $"m2"), "half", "m")
+ assert(maps.schema("m").dataType ===
+ MapType(StringType, IntegerType, valueContainsNull = true))
+ checkAnswer(
+ maps.select($"store", $"half", $"m".getItem("k").isNull),
+ Seq(
+ Row("north", "m1", false), Row("north", "m2", false),
+ Row("south", "m1", false), Row("south", "m2", true)))
+ }
+
+ test("unpivot keeps nested field names and metadata of the first value") {
+ val metadata1 = new MetadataBuilder().putString("comment", "first").build()
+ val metadata2 = new MetadataBuilder().putString("comment", "second").build()
+ // s1 and s2 differ only in the nested field name case, nullability and metadata
+ val schema = new StructType()
+ .add("id", IntegerType, nullable = false)
+ .add("s1", new StructType().add("f", IntegerType, nullable = false, metadata1))
+ .add("s2", new StructType().add("F", IntegerType, nullable = true, metadata2))
+ val df = spark.createDataFrame(
+ sparkContext.parallelize(Seq(Row(1, Row(1), Row(null)))), schema)
+
+ Seq(
+ Array[Column]($"s1", $"s2") -> new StructType().add("f", IntegerType, true, metadata1),
+ Array[Column]($"s2", $"s1") -> new StructType().add("F", IntegerType, true, metadata2)
+ ).foreach { case (values, expectedType) =>
+ val unpivoted = df.unpivot(Array($"id"), values, "col", "value")
+ assert(unpivoted.schema("value").dataType === expectedType)
+ // the values are not cast, since they have the same data type up to nullability
+ assert(!unpivoted.queryExecution.analyzed.exists(
+ _.expressions.exists(_.exists(_.isInstanceOf[Cast]))))
+ checkAnswer(
+ unpivoted.select($"col", $"value.f".isNull),
+ Seq(Row("s1", false), Row("s2", true)))
+ }
+ }
+
+ test("unpivot takes nested nullability of the first value with the legacy flag") {
+ withSQLConf(SQLConf.LEGACY_UNPIVOT_USE_FIRST_VALUE_NESTED_NULLABILITY.key -> "true") {
+ // a has NOT NULL elements, b has nullable elements
+ val df = Seq((1, Option.empty[Int])).toDF("id", "v")
+ .select($"id", array($"id").as("a"), array($"v").as("b"))
+ Seq(
+ Array[Column]($"a", $"b") -> false,
+ Array[Column]($"b", $"a") -> true
+ ).foreach { case (values, containsNull) =>
+ val unpivoted = df.unpivot(Array($"id"), values, "col", "value")
+ assert(unpivoted.schema("value").dataType === ArrayType(IntegerType, containsNull))
+ }
+ }
+ }
+
test("ORDER BY on a column dropped by UNPIVOT is rejected") {
// UNPIVOT drops the unpivoted value columns, so `ORDER BY t.a` cannot resolve against them;
// hidden-output insertion appends the column to an operator whose child does not produce it,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org