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 b94bbfb27b82 [SPARK-58089][SQL][FOLLOWUP] Preserve join extraction
semantics
b94bbfb27b82 is described below
commit b94bbfb27b825ef684814402e77f4d78a83cfc1b
Author: Wenchen Fan <[email protected]>
AuthorDate: Tue Aug 18 01:46:06 2026 +0800
[SPARK-58089][SQL][FOLLOWUP] Preserve join extraction semantics
### What changes were proposed in this pull request?
Follow-up to https://github.com/apache/spark/pull/57190.
This patch makes variant extraction hoisting respect `Expression.throwable`
when an extraction
actually crosses a join. Throwable extractions cross a join only when
cast-error deferral is
enabled. The join-crossing check mirrors alias placement through
pass-through Projects, so a Join
elsewhere beneath an operator such as a Filter does not unnecessarily block
hoisting.
### Why are the changes needed?
Moving an extraction below a Join can evaluate it on rows that the Join
later eliminates. For an
expression classified as throwable, that can expose an exception which the
original plan would
not raise. `Expression.throwable` is the shared expression-level contract
used by optimizer rules
that can change evaluation behavior, and this rule should respect it as
well.
The rule must also distinguish an extraction that crosses a Join from one
that is placed above an
intervening operator. Merely finding a Join somewhere in the subtree is too
broad.
### Does this PR introduce _any_ user-facing change?
No. This is a follow-up correction to an unreleased optimizer change.
### How was this patch tested?
Ran:
```bash
build/sbt 'sql/testOnly
org.apache.spark.sql.execution.datasources.PushVariantIntoScanSuite'
```
All 31 tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5).
Closes #57956 from cloud-fan/SPARK-58089-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/internal/SQLConf.scala | 3 +-
.../datasources/PullOutVariantExtractions.scala | 56 ++++++++++++++--------
.../datasources/PushVariantIntoScanSuite.scala | 12 +++++
3 files changed, 51 insertions(+), 20 deletions(-)
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 b9abc03aeda2..4192a9f91ff7 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
@@ -7031,7 +7031,8 @@ object SQLConf {
"per-row cast-error companion column (nullable string) so that the
cast error is only " +
"raised when the row is consumed by the user expression. Without this
flag, the cast is " +
"always evaluated and any failure raises immediately, even when the
surrounding " +
- "expression would not have consumed the failing row.")
+ "expression would not have consumed the failing row. This also allows
throwable variant " +
+ "extractions to be hoisted across joins while preserving their
original error timing.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.SESSION)
.booleanConf
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala
index 6695e24ad618..0ed27b12edb2 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala
@@ -107,18 +107,13 @@ import org.apache.spark.sql.types.VariantType
*
* 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.
+ * -- so a cast failure can surface for a row the un-hoisted plan would never
have cast. This is the
+ * same pre-existing trade-off as [[PushVariantIntoScan]] pushing casts below
a `Filter`. When
+ * [[SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR]] is set, the cast error
is deferred until the
+ * original expression consumes the failing row, suppressing errors from rows
eliminated by either
+ * operator. With the flag off, strict `VariantGet` and `Cast` retain their
existing behavior
+ * because they are not currently classified as [[Expression.throwable]]. If
an extraction is
+ * classified as throwable, it crosses a join only when cast-error deferral is
enabled.
*/
object PullOutVariantExtractions extends Rule[LogicalPlan] {
@@ -177,11 +172,18 @@ object PullOutVariantExtractions extends
Rule[LogicalPlan] {
case _ => false
}
+ private def isJoinHoistable(e: Expression): Boolean = {
+ // Unlike optimizer rules that must stop at throwable expressions, scan
pushdown can preserve
+ // the original error timing with its per-row cast-error companion column
when enabled.
+ isHoistable(e) && (!e.throwable ||
+ SQLConf.get.getConf(SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR))
+ }
+
/**
* Collects hoisted extractions as aliases, de-duplicated by canonical form
so a repeated
* extraction maps to a single output slot.
*/
- private class ExtractionHoister {
+ private class ExtractionHoister(hoistable: Expression => Boolean =
isHoistable) {
private val extracted = mutable.LinkedHashMap.empty[Expression, Alias]
def aliases: Seq[NamedExpression] = extracted.values.toSeq
@@ -194,10 +196,25 @@ object PullOutVariantExtractions extends
Rule[LogicalPlan] {
/** Replaces every hoistable extraction in `e` with a reference to its
(new) alias. */
def hoist(e: Expression): Expression = e.transformDown {
- case ex if isHoistable(ex) => aliasFor(ex)
+ case ex if hoistable(ex) => aliasFor(ex)
}
}
+ // Mirrors the traversal in pushSideAliases for one extraction. A Project is
pass-through only
+ // when the extraction resolves against its child; all other operators stop
alias placement.
+ private def extractionCrossesJoin(child: LogicalPlan, e: Expression):
Boolean = child match {
+ case _: Join => true
+ case Project(_, grandChild) if e.references.subsetOf(grandChild.outputSet)
=>
+ extractionCrossesJoin(grandChild, e)
+ case _ => false
+ }
+
+ private def extractionHoister(child: LogicalPlan): ExtractionHoister = {
+ new ExtractionHoister(e =>
+ if (extractionCrossesJoin(child, e)) isJoinHoistable(e) else
isHoistable(e)
+ )
+ }
+
// Recursively pushes hoisted extraction aliases down through a join tree
until each lands in a
// `Project` directly above a non-join child (the scan side, or the
`Filter`/`Project` chain above
// it that `PhysicalOperation` collapses). Hoisting an extraction into a
`Project` above a `Join`
@@ -299,7 +316,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
rightHoister: ExtractionHoister): Seq[NamedExpression] = {
projectList.map { e =>
e.transformDown {
- case ex if isHoistable(ex) =>
+ case ex if isJoinHoistable(ex) =>
if (ex.references.subsetOf(leftOutput)) {
leftHoister.aliasFor(ex)
} else if (ex.references.subsetOf(rightOutput)) {
@@ -312,7 +329,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
}
private def rewriteAggregate(agg: Aggregate): LogicalPlan = {
- val hoister = new ExtractionHoister
+ val hoister = extractionHoister(agg.child)
// Only hoist extractions that sit inside an aggregate function's
arguments (or filter). A
// top-level extraction in `aggregateExpressions` is a grouping-key
reference (grouping keys are
// already pulled out by `PullOutGroupingExpressions`); hoisting it would
leave the `Aggregate`
@@ -388,7 +405,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
private def rewriteSortUnderProject(
project: Project, projectList: Seq[NamedExpression], sort: Sort):
LogicalPlan = {
- val hoister = new ExtractionHoister
+ val hoister = extractionHoister(sort.child)
val newOrder = sort.order.map(hoister.hoist(_).asInstanceOf[SortOrder])
if (hoister.isEmpty) {
project
@@ -419,7 +436,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
// yielding a multi-slot shredded struct with a full-variant slot -- the
same shape as a
// Sort-under-Project whose `v` is also selected. See the class doc.
private def rewriteBareSort(sort: Sort): LogicalPlan = {
- val hoister = new ExtractionHoister
+ val hoister = extractionHoister(sort.child)
val newOrder = sort.order.map(hoister.hoist(_).asInstanceOf[SortOrder])
if (hoister.isEmpty) {
sort
@@ -442,6 +459,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
project: Project, projectList: Seq[NamedExpression], join: Join):
LogicalPlan = {
val leftOutput = join.left.outputSet
val rightOutput = join.right.outputSet
+ // Join-crossing eligibility is checked at the routing sites before
aliasFor is called.
val leftHoister = new ExtractionHoister
val rightHoister = new ExtractionHoister
@@ -451,7 +469,7 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
// join (its extractions -- e.g. aggregate arguments hoisted here by
`rewriteAggregate`, or a
// user's `SELECT variant_get(...)`), so the pushdown sees them below the
join.
val newCondition = join.condition.map(_.transformDown {
- case ex if isHoistable(ex) =>
+ case ex if isJoinHoistable(ex) =>
if (ex.references.subsetOf(leftOutput)) {
leftHoister.aliasFor(ex)
} else if (ex.references.subsetOf(rightOutput)) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
index d6ed274327a6..e92e938ad142 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
@@ -34,6 +34,18 @@ trait PushVariantIntoScanSuiteBase extends
SharedSparkSession {
// Whether the reader-deferral tests should exercise the V2 read path.
Subclasses override.
protected def useV2: Boolean
+ test("hoistable variant extractions are not classified as throwable") {
+ // If either expression becomes throwable, join hoisting requires
cast-error deferral.
+ val v = AttributeReference("v", VariantType)()
+ assert(!VariantGet(
+ v,
+ Literal("$.a"),
+ IntegerType,
+ failOnError = true,
+ timeZoneId = Some(localTimeZone)).throwable)
+ assert(!Cast(v, IntegerType, timeZoneId = Some(localTimeZone)).throwable)
+ }
+
// Write a parquet dataset via V1, then expose it as the temp view `T`. The
view's read path is
// V2 when `useV2`, V1 otherwise. Use this for tests that need to actually
execute a scan and
// compare V1 vs V2 behavior.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
