Skip to content

[SPARK-58089][SQL][FOLLOWUP] Preserve join extraction semantics #57956

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cloud-fan wants to merge 2 commits into
base: master
Choose a base branch
from
+60 −26
Open

[SPARK-58089][SQL][FOLLOWUP] Preserve join extraction semantics #57956

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b293ebe
[SPARK-58089][SQL][FOLLOWUP] Preserve join extraction semantics
cloud-fan Aug 12, 2026
c46ec50
[SPARK-58089][SQL] Use throwable metadata for join hoisting
cloud-fan Aug 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 31 additions & 26 deletions ...src/main/scala/org/apache/spark/sql/execution/datasources/PullOutVariantExtractions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,8 @@ package org.apache.spark.sql.execution.datasources

import scala.collection.mutable

import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, Cast, Expression,
NamedExpression, SortOrder}
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, AttributeSet, Cast,
Expression, NamedExpression, SortOrder}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
import org.apache.spark.sql.catalyst.plans.LeftExistence
Expand Down Expand Up @@ -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

@peter-toth peter-toth Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

* scan on rows the join later eliminates. Such extractions cross a join only when
* [[SQLConf.PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR]] is enabled. Expressions not classified as
* throwable retain the existing behavior, including strict `VariantGet` and `Cast` expressions.
*/
object PullOutVariantExtractions extends Rule[LogicalPlan] {

Expand Down Expand Up @@ -177,11 +167,16 @@ object PullOutVariantExtractions extends Rule[LogicalPlan] {
case _ => false
}

private def isJoinHoistable(e: Expression): Boolean = {

@qlong qlong Aug 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a reason to add this new gate? isJoinHoistable gate will not hoist a strict variant_get across a Join unless PUSH_VARIANT_INTO_SCAN_DEFER_CAST_ERROR is set. The ungated behavior is documented between line 108-122, and is more consistent with existing bahaviors. It seems this gate singles out variant for cross a join while allowing the other two cases to raise:

  1. variant_get below a Filter → raises
  2. cast(v as int) across a Join → raises (test @844, pre-existing base-PR behavior)
  3. variant_get across a Join → suppressed by the gate (test @881, new here)

I think it's good that all three have consistent behavior. We also have existing switches:

  • pushVariantIntoScan.pullOutExtractions = false to turn off the pull-out rule completely
  • pushVariantIntoScan.deferCastError = true to defer the error past eliminated rows uniformly for every strict extraction

@cloud-fan cloud-fan Aug 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. I replaced the VariantGet-specific gate with the generic Expression.throwable check. VariantGet and Cast are not currently classified as throwable, so this preserves the existing behavior for both while making the rule ready to honor that metadata if we classify them in the future. I also removed the new test that singled out strict variant_get.

@qlong qlong Aug 12, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 on using throwable, it is cleaner. Minor suggestion, please feel free to ignore. I am a bit worried there is no clear signal for hoist behavior change when someone classifies Cast and VariantGet as throwable in the future. Should we add a unit test to assert those two are throwable and add some comment there?

@peter-toth peter-toth Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Finding 4. @qlong's worry that there'd be no signal if Cast/VariantGet get classified as throwable later is well founded, and I think there's a sharper version of it: the flag isn't this rule's to define. CombineFilters (Optimizer.scala:2039) and PushPredicateThroughJoin (:2475, :2507) both branch on !cond.throwable, so nobody can set Cast.throwable = true or VariantGet.throwable = true in order to activate this gate -- that would change filter merging and join-condition pushdown for every query at the same time. So the hook can't really be honored on purpose for variant shredding, and in the other direction, if the flag is ever flipped for an unrelated reason this rule silently stops shredding across joins as a side effect. ExprUtils.scala:244-246 records a similar reservation for a different caller: "this deliberately does not rely on Expression.throwable, which is opt-in metadata that most expressions do not override."

That makes @qlong's suggested test worth more than a minor nit, and I'd also add a comment next to isJoinHoistable saying the check is inert today and what would make it fire, because right now it reads as a live guard. For the record on how inert it is: throwable defaults to children.exists(_.throwable) (Expression.scala:169) and Sequence (collectionOperations.scala:3479) is the only override in all of sql/, while isHoistable only admits an Attribute/GetStructField chain plus foldable Literals -- so isJoinHoistable is currently exactly isHoistable.

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
Expand All @@ -194,10 +189,14 @@ 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)
}
}

private def extractionHoister(crossesJoin: Boolean): ExtractionHoister = {
if (crossesJoin) new ExtractionHoister(isJoinHoistable) else new ExtractionHoister
}

// 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`
Expand Down Expand Up @@ -299,7 +298,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)) {
Expand All @@ -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))

@peter-toth peter-toth Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

// 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`
Expand Down Expand Up @@ -388,7 +387,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.containsPattern(JOIN))
val newOrder = sort.order.map(hoister.hoist(_).asInstanceOf[SortOrder])
if (hoister.isEmpty) {
project
Expand Down Expand Up @@ -419,7 +418,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.containsPattern(JOIN))
val newOrder = sort.order.map(hoister.hoist(_).asInstanceOf[SortOrder])
if (hoister.isEmpty) {
sort
Expand All @@ -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)

@peter-toth peter-toth Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

val rightHoister = extractionHoister(crossesJoin = true)

// A single variant extraction references exactly one attribute, hence one join side; route it
// to that side's hoister. Anything not cleanly on one side is left in place. We hoist from both
// the join condition (its extractions feed the equi-key comparison) and the Project above the
// 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)) {
Expand Down Expand Up @@ -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.

@qlong qlong Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 on nullablity fix

val newJoinOutput = AttributeMap(newJoin.output.map(attr => attr -> attr))

@peter-toth peter-toth Aug 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

val remappedProjectList = newProjectList.map(_.transformDown {
case attr: Attribute => newJoinOutput.getOrElse(attr, attr)
}.asInstanceOf[NamedExpression])
project.copy(projectList = remappedProjectList, child = newJoin)
}
}
}
29 changes: 29 additions & 0 deletions .../src/test/scala/org/apache/spark/sql/execution/datasources/PushVariantIntoScanSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import org.apache.spark.SparkConf
import org.apache.spark.sql.QueryTest
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.variant._
import org.apache.spark.sql.catalyst.plans.LeftOuter
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation
import org.apache.spark.sql.internal.SQLConf
Expand Down Expand Up @@ -776,6 +777,34 @@ trait PushVariantIntoScanSuiteBase extends SharedSparkSession {
}
}

test("left outer join: projected right-side attributes use widened nullability") {
val leftKey = AttributeReference("leftKey", IntegerType, nullable = false)()
val rightKey = AttributeReference("rightKey", IntegerType, nullable = false)()
val rightVariant = AttributeReference("rightVariant", VariantType, nullable = false)()
val join = Join(
LocalRelation(leftKey),
LocalRelation(rightKey, rightVariant),
LeftOuter,
Some(EqualTo(leftKey, rightKey)),
JoinHint.NONE)
val extraction = VariantGet(
rightVariant,
path = Literal("$.label"),
targetType = StringType,
failOnError = false,
timeZoneId = Some(localTimeZone))
val plan = Project(Seq(rightKey, Alias(extraction, "label")()), join)

PullOutVariantExtractions(plan) match {
case Project(projectList, rewrittenJoin: Join) =>
val projectedRightKey = projectList.head.asInstanceOf[Attribute]
val childRightKey = rewrittenJoin.output.find(_.exprId == rightKey.exprId).get
assert(projectedRightKey.nullable)
assert(projectedRightKey.nullable == childRightKey.nullable)
case other => fail(s"Expected a Project over Join, but got:\n$other")
}
}

test("full outer join: extractions on both (nullable) sides are pushed and value-preserving") {
// In a FULL OUTER join both sides are nullable: unmatched rows are null-padded on the opposite
// side. Extractions on each side push onto that side, computed before null-padding; the join
Expand Down