spark-test commented on code in PR #60061:
URL: https://github.com/apache/spark/pull/60061#discussion_r4488922777
##########
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))
+ }
+ if (widened == refTypes) {
+ cteDef
+ } else {
+ l.getTagValue(LOOP_ORIGIN) match {
+ case Some(child) =>
+ val restored = cteDef.copy(child = child)
+ restored.setTagValue(SELF_REFERENCE_TYPES, widened)
+ restored
+ case None =>
+ throw SparkException.internalError("The self-reference of
recursive CTE " +
+ s"${cteDef.id} needs wider types
${widened.map(_.sql).mkString(", ")}, but the " +
+ "loop has no origin to rebuild it from.")
+ }
+ }
+ case None => cteDef
+ }
+ }
+
+ // Primitive Scala UDFs only become nullable in the later "ScalaUDF Null
Handling" batch. Applies
+ // that batch to a copy, so the self-reference also covers the NULLs those
UDFs write.
+ private def withUdfNullHandling(plan: LogicalPlan): LogicalPlan = {
+ if (plan.containsPattern(SCALA_UDF)) {
+ val handled = UpdateAttributeNullability(HandleNullInputsForUDF(plan))
Review Comment:
**Non-blocking (P2):** `UpdateAttributeNullability` only refreshes the
top-level `nullable` flag of attribute references, and an attribute keeps the
data type it was resolved with. When the array is built one operator below the
column's output, the copy therefore still declares non-nullable elements after
`HandleNullInputsForUDF` wraps the UDF, and the self-reference is not widened.
With `spark.udf.register("f", (x: Long) => x + 1)`:
```sql
WITH RECURSIVE t(lvl, arr, was_null) AS (
SELECT 0, ARRAY(id), false FROM range(200)
UNION ALL
SELECT lvl + 1, a, arr[0] IS NULL
FROM (SELECT lvl, arr, ARRAY(f(IF(lvl = 0, CAST(NULL AS BIGINT), arr[0])))
AS a FROM t)
WHERE lvl < 2
)
SELECT lvl, count_if(was_null) FROM t GROUP BY lvl ORDER BY lvl
```
Level 1 writes `[null]` for all 200 rows, but level 2 reads them through the
narrower self-reference. Generated `GetArrayItem` code skips `isNullAt` when
`containsNull` is false, so level 2 counts 0 instead of 200. Moving
`ARRAY(f(...))` into the top projection gives 200, so the comment above, which
says the self-reference covers the NULLs those UDFs write, overstates the
coverage. An anchor that builds the array in a subquery, with a recursion that
passes it through, fails the same way. Steps small enough to become local
relations hide the problem, because interpreted evaluation checks `isNullAt`.
This extends existing discussions
[1](https://github.com/apache/spark/pull/60061#issuecomment-6664983350),
[2](https://github.com/apache/spark/pull/60061#pullrequestreview-5779476661).
**Recommended change:** After applying Scala UDF null handling to the anchor
and recursion copies, refresh each attribute reference in the copies to its
producer's data type when that type differs only by more nullable nested
positions, and repeat until nothing changes, before the widening reads the
copies' outputs.
**Why this works:** Refreshing references bottom-up carries the nullable
element, value or field type that a wrapped UDF introduces in a FROM subquery
or inner projection through every operator above it. The copy's output for the
column then declares the NULL, and the merge widens the self-reference. Only
nullability changes, so the copies keep their structure and the widening stays
monotone.
**Scope:** Make the widening's UDF-handled copies propagate nested
nullability through attribute references, and cover the subquery shapes with
UDF tests.
**Risks:** Columns that copy or pass through a nested value built around a
primitive Scala UDF below the column's output operator now get nullable nested
result types, so persisted views and NOT NULL casts over them can fail like
those over other widened columns.
**Constraints:** The refresh must run only on the copies used to compute
types and only widen nullability. The live plan then still receives UDF null
handling exactly once, and the widening still terminates.
**Exclusions:**
- The main query's reads of UDF-written nested values through the CTE result
keep their stale types; that general analyzer behavior stays outside this
change.
**Success:** When the anchor or the recursion builds an array, map or struct
around a primitive Scala UDF that returns NULL, in any operator below the one
that outputs the column, the self-reference declares that position nullable and
a later iteration reads the NULL back as NULL.
**Verification:**
- **Regression:** A recursion that builds ARRAY(f(...)) around a primitive
Long => Long UDF in a FROM subquery and outputs it as the column reads the
UDF's NULL back as NULL at the next level, on materialized steps with generated
code.
- **Regression:** An anchor that builds the array around such a UDF in a
subquery, with a recursion that passes the column through, reads the UDF's
NULLs back as NULL at the next level.
- **Compatibility:** Recursions without Scala UDFs, and UDF cases whose
nested value is built in the top projection, keep the same self-reference and
result types as before the change.
##########
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) =>
Review Comment:
**Blocking (P1):** This merge reads only declared nullability, and
`rewriteRecursiveCTERefs` still declares every self-reference column nullable
(lines 336-341). A value built from self-reference columns, such as `ARRAY(n)`
or `named_struct('lvl', n)`, is therefore typed as holding NULLs even when it
never can, and the widening carries that into every column that copies it or
derives from it. For example:
```sql
CREATE VIEW v AS WITH RECURSIVE t(n, a, b) AS (
SELECT 1, ARRAY(1), ARRAY(1)
UNION ALL
SELECT n + 1, ARRAY(n), a FROM t WHERE n < 3
) SELECT * FROM t
```
`n` starts at 1 and only grows, so no NULL is ever written. On the base
branch, `b` is `ARRAY<INT>` with non-nullable elements, and the view stores
that type. With this change, the first round widens the self-reference of `a`,
because `ARRAY(n)` is declared with `containsNull = true`, and the second round
widens `b`, which copies `a`. Reading the view then wraps `b` in an ANSI `Cast`
(SCHEMA COMPENSATION, the default) or an `UpCast` (SCHEMA BINDING) to the
stored type. Both reject going from nullable to non-nullable elements, so
`SELECT * FROM v` stops resolving although its rows were correct. Casts of such
columns to NOT NULL nested types fail the same way, and no setting restores the
old types. The migration note describes only recursions that write a NULL, so
it does not point these users at the cause. This also contradicts the PR
description's statement that output types only change for columns that can hold
such NULLs.
A path-tracking recursion whose `prev_path` copies `path =
array_append(path, n + 1)` fails the same way for a different reason. Spark
types every `array_append` result with nullable elements
(`ArrayInsert.dataType` is `first.dataType.asNullable`), and it does the same
for element reads such as `arr[0]`.
This extends existing discussions
[1](https://github.com/apache/spark/pull/60061#issuecomment-6664983350),
[2](https://github.com/apache/spark/pull/60061#pullrequestreview-5779476661).
**Recommended change:** Derive the top-level nullability of each
self-reference column from what the anchor and the recursion can produce, using
the same widen-and-rebuild fixed point that already widens nested types,
instead of always declaring it nullable. Add a TODO at the widening noting that
copies of values Spark types as nullable whatever their inputs, such as
array_append results, still widen.
**Why this works:** Each self-reference column starts at the anchor's
top-level nullability. Like the nested types, it widens whenever the
recursion's output for that column is nullable, and the loop is rebuilt through
the existing restore path. A value whose nested nullability Spark derives from
a column that can never be NULL, such as ARRAY(n) or named_struct('lvl', n), is
then declared with non-nullable elements or fields. It no longer widens its own
self-reference or the columns that copy it, while any column that the anchor or
the recursion can make NULL still ends up nullable.
**Scope:** Extend the self-reference widening in the recursive CTE analyzer
rule to top-level nullability, and cover it with analyzer and SQL tests.
**Compatibility:** Columns the recursion only passes through keep the nested
nullability the anchor can produce, and every NULL that the anchor or recursion
writes is still read back as NULL.
**Risks:** When neither the anchor nor the recursion can produce NULL for a
self-reference column, that column, and the CTE result columns derived from it,
become non-nullable at the top level. That changes plans for those recursive
CTEs, with fewer null checks and simplified IS NULL predicates, and the
nullability that new persisted views record; this ships with the fix and has no
separate switch.
**Constraints:** The top-level widening must start from the anchor's
nullability, only ever add nullability, and rebuild through the same restore
path. It then terminates, and every NULL a later iteration can write stays
covered, including NULLs from outer joins and conditional expressions.
**Exclusions:**
- Values that Spark declares nullable whatever the self-reference's
nullability, such as element reads like arr[0] or m['k'] and the elements of
every array_append, array_prepend or array_insert result, still widen the
columns that copy them, including in recursions that can never write a NULL.
Avoiding that needs those expressions typed precisely, which is a separate
change; the TODO and the migration note name it.
- Persisted views and NOT NULL casts over columns that can actually receive
such NULLs still fail as documented; that trade-off remains the open decision
about views and casts.
**Success:** A recursion whose self-reference column can never be NULL keeps
that column non-nullable; nested values whose nullability Spark derives from
it, such as ARRAY(n) or named_struct('lvl', n), keep non-nullable element and
field types, and columns that copy those values keep the base branch's result
types. Every NULL that the anchor or the recursion can write, at the top level
or in a nested position, is still covered by the self-reference's declared
types and is read back as NULL.
**Verification:**
- **Regression:** Take a recursive CTE whose copied column is built only
from self-reference values that can never be NULL, such as ARRAY(n) or
named_struct('lvl', n) with n counting up from 1, and one whose column derives
its nested type from such a copy, such as TRANSFORM(a, x -> x + 1). The copying
and deriving columns keep the base branch's non-nullable nested result types,
and a persisted view created with those types still reads.
- **Behavior:** A recursion that writes a NULL into a nested position,
directly or into a column that another column copies, still widens the
self-reference and the copying columns, and later iterations read the NULL back
as NULL on both materialized and local steps.
- **Behavior:** A self-reference column whose recursion produces NULL only
in a later round, such as CASE WHEN n > 5 THEN NULL ELSE n + 1 END, converges
to a nullable self-reference and reads that NULL back as NULL.
- **Compatibility:** Persisted views created on the base branch over
recursive CTEs that never write a NULL keep resolving, and columns the
recursion only passes through keep the nested nullability the anchor can
produce.
##########
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:
**Nit (P3):** `ARRAY(arr[0] + 1)` is already declared with nullable
elements, because `GetArrayItem` with a constant ordinal over anything other
than a `CreateArray` is always nullable. The recursion's own output therefore
widens `arr`, and this case still returns `Row(1, 100L)` if the widening stops
applying UDF null handling to the anchor. A case whose recursion passes `arr`
through, such as `SELECT lvl + 1, arr, arr[0] IS NULL FROM t WHERE lvl < 1`
over the same anchor, depends on the anchor half alone and would catch that
regression.
**Verification:**
- **Regression:** Use an anchor that writes NULLs through a primitive Long
=> Long UDF, with 200-row materialized steps, and a recursion that passes the
array through. The next level reads those NULLs back as NULL, and the case
fails if the anchor's UDF null handling is dropped from the widening.
##########
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:
**Nit (P3):** This entry doesn't match which columns actually change type:
- The column the recursion writes a NULL into keeps the type it had.
`UnionLoop` output already merged the recursion's declared nested nullability,
so a column written with a NULL literal was already nullable on the base
branch. A column written by a primitive Scala UDF stays non-nullable, because
the query's reference to the CTE is fixed before UDF null handling runs.
- The columns that change are the ones that copy, or derive their nested
nullability from, a widened self-reference column. The widening follows
declared nullability, so they change even when the recursion never writes a
NULL, for example `b` in `SELECT n + 1, ARRAY(n), a FROM t`.
- Columns that the recursion only passes through do change type when a
primitive Scala UDF in the anchor builds them.
A user whose view or cast fails after upgrading is pointed at the wrong
column, or may conclude that the entry doesn't apply. Naming the copy and
derived columns, including copies of values that Spark always types as nullable
such as `array_append` results, and the UDF-built pass-through columns would
fix that.
**Verification:**
- **Inspection:** Each kind of column the entry names as changed actually
changes type in the analyzer: copies and derived columns of a self-reference
column that can hold NULL, and pass-through columns that primitive Scala UDFs
in the anchor build. Each kind it names as unchanged, the written column and
other pass-through columns, keeps its base-branch result type.
##########
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:
**Nit (P3):** `c.output` doesn't depend on `i`, but it is evaluated once per
column, and `Project.output` rebuilds `projectList.map(_.toAttribute)` on every
call. For an n-column CTE, each check therefore builds about 2n outputs of n
attributes instead of two, on every remaining Resolution pass. Reading each
child's output once before the loop, for example `val childOutputs =
children.map(_.output.toIndexedSeq)`, gives the same types in the same merge
order.
**Verification:**
- **Behavior:** Every existing widening case still produces the same
self-reference and result types: pass-through, two-round copy, inner WITH,
column list and Scala UDF.
- **Inspection:** The widening reads each child's output once per check
rather than once per column.
--
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]
