peter-toth commented on code in PR #57956:
URL: https://github.com/apache/spark/pull/57956#discussion_r3770985741
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala:
##########
@@ -105,20 +105,10 @@ import org.apache.spark.sql.types.VariantType
* [[SQLConf.PUSH_VARIANT_INTO_SCAN]] is also enabled) and only fires when a
hoistable extraction is
* present, so non-variant plans are untouched.
*
- * Cast-error surface: relocating a strict extraction (`variant_get(...,
failOnError = true)` or a
- * strict `Cast`) below a `Join` means it is evaluated at the scan on rows the
join later eliminates
- * -- so a cast failure can surface for a row the un-hoisted plan would never
have cast (here the
- * eliminating rows come from the *other* joined table). This is the same
pre-existing trade-off as
- * [[PushVariantIntoScan]] pushing casts below a `Filter`, not a new error
class: in both, the
- * strict cast runs before the operator that would have discarded the failing
row. This rule only
- * relocates the extraction into a `Project`; [[PushVariantIntoScan]] still
does the scan-level
- * materialization and, when
[[SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR]] is set (default
- * false), wraps the cast with a per-row cast-error companion slot so the
error is only raised when
- * the original expression consumes the failing row. That deferral is
provenance-agnostic -- it acts
- * on the relocated extraction regardless of how it reached the `Project` --
so enabling the flag
- * suppresses the join-eliminated-row error exactly as it does the
filter-eliminated-row case. With
- * the flag off (the default), the strict cast raises immediately on any
failing scanned row, as
- * documented for below-`Filter` pushdown.
+ * Cast-error surface: relocating a throwable extraction below a `Join` means
it is evaluated at the
Review Comment:
**Finding 2.** This paragraph states a rule and then withdraws it: "Such
extractions cross a join only when [deferral] is enabled", immediately followed
by "Expressions not classified as throwable retain the existing behavior,
including strict `VariantGet` and `Cast` expressions" -- which is every
extraction this rule hoists. Someone coming here for the cast-error contract
finds a sentence that applies to nothing.
The paragraph it replaced was written to answer your own review question on
#57190, and it is still accurate now that behavior is back to base: it
explained *why* relocating a strict extraction below a `Join` is the same
accepted trade-off as `PushVariantIntoScan` pushing a cast below a `Filter`,
and how `PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR` interacts with it. Deleting
it loses the record of that decision, which is the thing a future reader will
come looking for. I'd restore it and add a sentence saying the `throwable`
check is currently inert and what would make it fire.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala:
##########
@@ -312,7 +311,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
}
private def rewriteAggregate(agg: Aggregate): LogicalPlan = {
- val hoister = new ExtractionHoister
+ val hoister = extractionHoister(agg.child.containsPattern(JOIN))
Review Comment:
**Finding 6.** If the gate is ever made live, this isn't quite the right
question. `containsPattern(JOIN)` is true for a join anywhere in the subtree,
but whether the alias actually crosses a join is decided later, by
`pushSideAliases`: when the join sits under a `Filter`, the `case other` arm at
`:281` parks the alias in a `Project` above the `Filter` and it never crosses
the join -- yet this check classifies it as join-crossing. Same at
`rewriteSortUnderProject:390` and `rewriteBareSort:421`.
Not worth changing while the gate is inert, but worth a comment if the
machinery is being kept for later.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala:
##########
@@ -442,16 +441,16 @@ object PullOutVariantExtractions extends
Rule[LogicalPlan] {
project: Project, projectList: Seq[NamedExpression], join: Join):
LogicalPlan = {
val leftOutput = join.left.outputSet
val rightOutput = join.right.outputSet
- val leftHoister = new ExtractionHoister
- val rightHoister = new ExtractionHoister
+ val leftHoister = extractionHoister(crossesJoin = true)
Review Comment:
**Finding 5.** `crossesJoin = true` has no effect here.
`leftHoister`/`rightHoister` are only used via `aliasFor`, `aliases` and
`isEmpty` -- never `hoist` -- so the `hoistable` predicate they are constructed
with is never consulted. The gating on this path is done inline instead: at
`:453` for the join condition and at `:301` inside
`hoistProjectListExtractions`.
Passing the predicate reads as if the hoister enforced it, which will
mislead the next reader. Either construct plain `new ExtractionHoister` here,
or route those two call sites through `hoist` so there is a single place that
decides.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala:
##########
@@ -481,7 +480,13 @@ object PullOutVariantExtractions extends Rule[LogicalPlan]
{
left = pushSideAliases(join.left, leftHoister.aliases, needed),
right = pushSideAliases(join.right, rightHoister.aliases, needed),
condition = newCondition)
- project.copy(projectList = newProjectList, child = newJoin)
+ // Outer joins widen attributes on their nullable side. Remap references
in the parent
+ // Project to the copied Join's actual output so their nullability
matches the child.
+ val newJoinOutput = AttributeMap(newJoin.output.map(attr => attr ->
attr))
Review Comment:
**Finding 1.** I measured this one, because it contradicts your own
verification on #57190.
**What I ran.** At `c46ec508` I reverted only this remap (back to
`project.copy(projectList = newProjectList, child = newJoin)`) and reran
`PushVariantIntoScanSuite` and `PushVariantIntoScanV2Suite`: **108 of 110 tests
pass, and the only 2 failures are the new test itself** -- it lives in the
shared base trait, so it runs once per suite. Its failure is
`projectedRightKey.nullable was false` at `PushVariantIntoScanSuite.scala:802`.
I also added a throwaway check over six real SQL queries -- LEFT / RIGHT /
FULL OUTER joins, the `Project`-over-`Join` shape, `Sort`-over-`Join` and
`Aggregate`-over-`Join` -- asserting that every attribute reference in every
node agrees with that child's output on nullability, evaluated on the analyzed
plan, on this rule's own output, and on the fully optimized plan. With the
remap reverted: zero violations on all six, including the direct SQL analogue
of the new test, `select d.k, variant_get(d.data, '$.label', 'string') from SS
ss left join DIM d on ss.k = d.k`.
**Why there is nothing to fix.** No alias this rule creates can be
non-nullable, so the outer join's widening has nothing to widen:
`VariantGet.nullable` is `true` unconditionally
(`variantExpressions.scala:446`), and `Cast.nullable` is `child.nullable ||
Cast.forceNullable(...)` (`Cast.scala:773`) with
`Cast.forceNullable(VariantType, _) = true` (`Cast.scala:551`). And
`pushSideAliases` only adds and drops columns, never rebuilding a retained
attribute, so `newJoin.output` agrees with `join.output` on nullability for
everything `newProjectList` references.
That is the conclusion your #57190 review already recorded:
"`VariantGet.nullable` and `Cast(VariantType->_)` are always `nullable=true`,
and `Join.computeOutput` widens the null-supplying side to nullable, so every
parent reference to a pushed `_ve` is nullable-over-nullable (AGREES) -- no
plan-integrity nullability mismatch." I think that reading was right.
**On the test** (`PushVariantIntoScanSuite.scala:796`). It builds
`Project(Seq(rightKey /* nullable = false */, ...), <LeftOuter join>)`. The
analyzer resolves a `Project`'s references against `child.output`, and
`Join.computeOutput` maps the nullable side through `withNullability(true)`
(`basicLogicalOperators.scala:703`), so a `Project` above a `LeftOuter` join
never holds the pre-widening `rightKey`. I don't doubt "verified to fail before
the production fix and pass after it" -- it does fail. It fails because the
test constructs the inconsistency itself, which is why it isn't evidence of a
defect.
So either there is a plan shape I haven't found -- in which case a test on a
real query would be worth far more than this one -- or the remap is
normalization rather than a fix. If it's normalization I don't object to
keeping the code, but I'd drop or relabel the test and the description claim:
as written, someone will later read this as a shipped bug fix and may backport
it to branch-4.3 / branch-4.x for nothing. And if it is kept, the same
normalization is missing in the sibling paths -- `rewriteAggregate`'s
`newAggExprs`, `newOrder` in both `Sort` paths, and `pushSideAliases:275` --
which makes the intent hard to read.
--
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]
