stevomitric commented on code in PR #60061:
URL: https://github.com/apache/spark/pull/60061#discussion_r4502098977
##########
docs/sql-migration-guide.md:
##########
@@ -67,6 +67,7 @@ license: |
- 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 other columns are still dropped, as the concatenation invalidates them,
and a join that reduced the partition keys onto a common key space (see
`spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled`) reports no
order, since the merged partitions then share only the reduced key. This can
remove a `Sort` downstream of the merge. To restore the previous behavior, set
`spark.sql.sources.v2.bucketing.preserveKeyOrderingOnCoalesce.enabled` to
`false`.
- Since Spark 4.4, when the `from_protobuf` option
`unwrap.primitive.wrapper.types` is enabled, `from_protobuf` unwraps a present
Protobuf primitive wrapper (such as `google.protobuf.Int32Value` or
`StringValue`) whose inner `value` is unset or set to its default to the type's
default value (`0`, `false`, `""`, or empty binary), regardless of the
`emit.default.values` option. Previously such a wrapper unwrapped to `NULL`
unless the `emit.default.values` option was enabled, which could also place a
`NULL` into a non-nullable array element or map value. An absent wrapper field
still unwraps to `NULL`.
- 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` may have recorded the non-nullable nested type in its
schema and can fail to read; recreate it with `CREATE OR REPLACE VIEW`. To
restore the previous behavior, set
`spark.sql.legacy.unpivot.useFirstValueNestedNullability` to `true`.
+- Since Spark 4.4, a recursive CTE whose recursive term writes a `NULL` into
an array element, map value or struct field of a column that the anchor built
as non-nullable returns that `NULL`. Previously, the next iteration read such a
value through the anchor's non-nullable nested type and could return a
non-`NULL` value such as `0`, with `IS NULL` on it returning `false`, or fail
with a `NullPointerException`. The nested type of such a column, and of columns
that copy it, is now nullable in the result, so casting it to a type with a
`NOT NULL` nested type fails. A view created by an earlier Spark version over
such a recursive CTE may have recorded the non-nullable nested type in its
schema and can fail to read; recreate it with `CREATE OR REPLACE VIEW`, or run
`ALTER VIEW ... WITH SCHEMA EVOLUTION`. Columns that the recursive term only
passes through keep their type.
Review Comment:
done
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveWithCTE.scala:
##########
@@ -303,22 +323,78 @@ object ResolveWithCTE extends Rule[LogicalPlan] {
// Assumes that `anchor` is already resolved. If `columnNames` is set (which
happens if an
// UnresolvedSubqueryColumnAliases is atop a Union) also places an
UnresolvedSubqueryColumnAliases
// node above UnionLoopRef. At some later stage
UnresolvedSubqueryColumnAliases gets resolved to
- // a Project node.
+ // a Project node. `refTypes` overrides the anchor's data types for the
UnionLoopRef output.
private def rewriteRecursiveCTERefs(
recursion: LogicalPlan,
anchor: LogicalPlan,
cteDefId: Long,
- columnNames: Option[Seq[String]]) = {
+ columnNames: Option[Seq[String]],
+ refTypes: Option[Seq[DataType]] = None) = {
+ val types = refTypes.getOrElse(anchor.output.map(_.dataType))
recursion.transformUpWithSubqueriesAndPruning(_.containsPattern(CTE)) {
case r: CTERelationRef if r.recursive && r.cteId == cteDefId =>
// We mark the output of UnionLoopRef as nullable as it may become
NULL in the following
// iterations.
- val ref =
- UnionLoopRef(r.cteId,
anchor.output.map(_.newInstance().withNullability(true)), false)
+ val output = anchor.output.zip(types).map { case (a, dt) =>
+ a.newInstance().withNullability(true).withDataType(dt)
+ }
+ val ref = UnionLoopRef(r.cteId, output, false)
columnNames.map(UnresolvedSubqueryColumnAliases(_, ref)).getOrElse(ref)
}
}
+ // The anchor's types, unless widenSelfReference has found wider ones.
+ private def selfReferenceTypes(cteDef: CTERelationDef, anchor: LogicalPlan):
Seq[DataType] =
+
cteDef.getTagValue(SELF_REFERENCE_TYPES).getOrElse(anchor.output.map(_.dataType))
+
+ // A recursion can write NULLs into nested types that the self-reference
declares non-nullable,
+ // e.g. ARRAY(CAST(NULL AS INT)) for an ARRAY(1) anchor. Once the loop is
resolved, widens the
+ // nullability of such self-reference types to the anchor's and recursion's
and rebuilds the
+ // loop from the original definition. This repeats until no type widens.
+ private def widenSelfReference(cteDef: CTERelationDef): CTERelationDef = {
+ val loop = if (cteDef.hasSelfReferenceAsUnionLoopRef) {
+ cteDef.child.collectFirst { case l: UnionLoop if l.id == cteDef.id => l }
+ } else {
+ None
+ }
+ loop.filter(_.resolved) match {
+ case Some(l) =>
+ val refTypes = cteDef.child.collectFirstWithSubqueries {
+ case r: UnionLoopRef if r.loopId == cteDef.id =>
r.output.map(_.dataType)
+ }.get
+ val children = Seq(l.anchor, l.recursion).map(withUdfNullHandling)
+ val widened = refTypes.zipWithIndex.map { case (dt, i) =>
+ children.foldLeft(dt)((t, c) => StructType.unionLikeMerge(t,
c.output(i).dataType))
Review Comment:
done
##########
sql/core/src/test/scala/org/apache/spark/sql/UDFSuite.scala:
##########
@@ -480,6 +480,42 @@ class UDFSuite extends SharedSparkSession {
}
}
+ test("SPARK-60875: NULLs from a primitive Scala UDF in a recursive CTE are
read back as NULL") {
+ spark.udf.register("f", (x: Long) => x + 1)
+ // Steps of 200 rows are materialized, so the recursion reads them through
generated code.
+ def nullsPerLevel(anchor: String, recursion: String): DataFrame = sql(
+ s"""WITH RECURSIVE t(lvl, arr, was_null) AS (
+ | $anchor
+ | UNION ALL
+ | $recursion
+ |)
+ |SELECT lvl, count_if(was_null) FROM t GROUP BY lvl ORDER BY
lvl""".stripMargin)
+
+ // The anchor writes the NULLs.
+ checkAnswer(
+ nullsPerLevel(
+ "SELECT 0, ARRAY(f(IF(id % 2 = 0, CAST(NULL AS BIGINT), id))), false
FROM range(200)",
+ "SELECT lvl + 1, ARRAY(arr[0] + 1), arr[0] IS NULL FROM t WHERE lvl <
1"),
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
