cloud-fan commented on code in PR #59267:
URL: https://github.com/apache/spark/pull/59267#discussion_r4284981637
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AsOfJoinResolver.scala:
##########
@@ -124,11 +132,62 @@ class AsOfJoinResolver(
)
cteRegistry.currentScope.tryPutWithCTE(
unresolvedOperator = unresolvedAsOfJoin,
- resolvedOperator = resolvedJoin
+ resolvedOperator = resolvedJoinWithOriginalOutput
)
}
}
+ /**
+ * If the join uses a hidden column, wraps it in a [[Project]] that also
keeps the hidden columns.
+ */
+ private def retainOriginalOutputIfHiddenColumnIsReferenced(
+ unresolvedAsOfJoin: AsOfJoin,
+ resolvedJoin: AsOfJoin,
+ leftNameScope: NameScope,
+ rightNameScope: NameScope): LogicalPlan = {
+ val originalOutput = AsOfJoin.computeOutput(
+ joinType = resolvedJoin.joinType,
+ leftOutput = leftNameScope.output,
+ rightOutput = rightNameScope.output
+ )
+ val originalOutputSet = AttributeSet(originalOutput)
+
+ if (resolvedJoin.references.subsetOf(originalOutputSet)) {
+ resolvedJoin
+ } else {
+ val qualifiedAccessOnlyColumns = AsOfJoin.computeOutput(
+ joinType = resolvedJoin.joinType,
+ leftOutput = leftNameScope.hiddenOutput.filter(_.qualifiedAccessOnly),
+ rightOutput = rightNameScope.hiddenOutput.filter(_.qualifiedAccessOnly)
+ )
+
+ operatorResolutionContextStack.current.baseOperator = Some(resolvedJoin)
+ Project(
+ withQualifiedAccessOnlyColumns(
Review Comment:
**Nit (P3):** In the single-pass analyzer, this appends the
qualified-access-only hidden columns whenever the AsOfJoin itself is not tagged
`TOP_LEVEL_OPERATOR`, and only the plan root gets that tag
(`Resolver.scala:230`). The helper's scaladoc relies on `PruneMetadataColumns`
to drop the unused ones. That pass starts from an empty needed set, and generic
operators forward only their own references (`PruneMetadataColumns.scala:49,
127`). A Project reached with an empty set keeps its whole list (lines 97-111).
So when every operator between the root and the wrapper references nothing,
`d.k` survives:
```sql
FROM using_left l ASOF JOIN (using_right r LEFT JOIN using_keys d USING (k))
MATCH_CONDITION (l.lts >= d.k)
LIMIT 1
```
The root here is `GlobalLimit(LocalLimit(...))`. Single-pass outputs `(k,
lts, k, rts, k)`, while fixed-point outputs `(k, lts, k, rts)`, the
FVT-ASOF-2-040 shape. `FROM (<same join>) AS s` and bare-`FROM` `UNION ALL`
operands behave the same way. Under dual-run,
`HybridAnalyzer.validateLogicalPlans` fails on the schema mismatch. With
single-pass alone, the extra column reaches the result. The fixed-point path
returns the right columns.
The JoinResolver and ASOF USING Projects already have this append-then-prune
gap at the merge base, for example `FROM a JOIN b USING (k) LIMIT 1`. So the
root cause is `PruneMetadataColumns`' root handling rather than this call site,
and this PR makes it reachable from `MATCH_CONDITION`. This answers the earlier
aliased-bare-`FROM` question; I confirmed it by tracing the source, not by
running it.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59267#pullrequestreview-5518239445).
**Recommended change:** Make PruneMetadataColumns seed its traversal at each
plan root with the ExprIds of the root's output attributes that are not
qualified-access-only, instead of an empty set. A Project reached through
reference-free pass-through ancestors then prunes the appended
qualified-access-only columns that no ancestor needs, the same as one reached
through a Project or Aggregate parent. Add SQL golden coverage to the ASOF JOIN
and USING join test families for bare-FROM shapes under LIMIT/OFFSET and a root
alias.
**Why this works:** Under TOP_LEVEL_OPERATOR, producers never append
qualified-access-only columns to the root itself. Per the PruneMetadataColumns
doc, any qualified-access-only entry in a Project list is an artificial append,
and the root output's user-visible columns are never qualified-access-only.
Seeding the needed set with the root's visible output ExprIds therefore keeps
every user-visible column and lets the non-empty-set branch drop each
artificial append that no operator above references. Generic pass-through
operators keep forwarding the parent set plus their own references, so the
wrapper below GlobalLimit, SubqueryAlias, or Union sees a non-empty set without
d.k and prunes it, matching fixed-point's Project(node.output).
**Scope:** Single-pass metadata-column pruning learns to recognize the real
root. ASOF JOIN and USING join golden tests gain bare-FROM shapes with
reference-free ancestors.
**Compatibility:** User-visible root columns, the non-empty-needed-set
pruning behavior for Project and Aggregate parents, and every fixed-point plan.
**Risks:** Single-pass plans change for existing NATURAL/USING joins under
reference-free roots. These currently disagree with fixed-point, so the change
restores parity, but any golden that happened to encode the leaked column would
need regeneration and review. A root output that is legitimately
qualified-access-only would be pruned. Per the PruneMetadataColumns doc this
should not happen, because resolution strips the flag from selected columns,
but the seed must not drop non-qualified-access-only root columns.
**Constraints:** Fixed-point behavior and the results of the default path
must not change. Existing single-pass plans that already match fixed-point,
including every FVT-ASOF-2-027..041 golden, must stay identical. Pruning must
stay a single top-down pass over each root with no extra plan traversals.
**Success:** `FROM using_left l ASOF JOIN (using_right r LEFT JOIN
using_keys d USING (k)) MATCH_CONDITION (l.lts >= d.k) LIMIT 1` and `FROM
(<same ASOF JOIN>) AS s` produce the schema (k, lts, k, rts) in single-pass,
and dual-run reports no schema or plan mismatch. A bare-FROM USING join under
LIMIT produces the same single-pass schema as fixed-point. Queries whose
ancestors reference columns keep their current single-pass plans.
**Verification:**
- **Regression:** Under dual-run, an ASOF JOIN with a hidden MATCH_CONDITION
operand in a bare FROM with LIMIT, and the same join under a root table alias,
resolve with identical single-pass and fixed-point schemas and plans, and the
hidden key does not appear in the output.
- **Regression:** Under dual-run, a bare-FROM NATURAL or USING join under
LIMIT or OFFSET produces a schema that matches fixed-point.
- **Compatibility:** Regenerated ASOF JOIN and NATURAL/USING join goldens
for queries with referencing ancestors are byte-identical to the current ones.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/AsOfJoinResolver.scala:
##########
@@ -124,11 +132,62 @@ class AsOfJoinResolver(
)
cteRegistry.currentScope.tryPutWithCTE(
unresolvedOperator = unresolvedAsOfJoin,
- resolvedOperator = resolvedJoin
+ resolvedOperator = resolvedJoinWithOriginalOutput
)
}
}
+ /**
+ * If the join uses a hidden column, wraps it in a [[Project]] that also
keeps the hidden columns.
+ */
+ private def retainOriginalOutputIfHiddenColumnIsReferenced(
+ unresolvedAsOfJoin: AsOfJoin,
+ resolvedJoin: AsOfJoin,
+ leftNameScope: NameScope,
+ rightNameScope: NameScope): LogicalPlan = {
+ val originalOutput = AsOfJoin.computeOutput(
+ joinType = resolvedJoin.joinType,
+ leftOutput = leftNameScope.output,
+ rightOutput = rightNameScope.output
+ )
+ val originalOutputSet = AttributeSet(originalOutput)
+
+ if (resolvedJoin.references.subsetOf(originalOutputSet)) {
+ resolvedJoin
+ } else {
+ val qualifiedAccessOnlyColumns = AsOfJoin.computeOutput(
Review Comment:
**Nit (P3):** Under `LEFT ASOF JOIN`, this `computeOutput` call is the only
thing that makes the appended hidden right-side columns nullable, and no
current test exercises it. FVT-ASOF-2-041's `d.k` comes from a nested `LEFT
JOIN`, so it is nullable already. No parent reads it either, so
`PruneMetadataColumns` removes it from the wrapper (the golden shows `Project
[k, lts, k, rts]`). Every case that reads a hidden key above the join is an
inner ASOF JOIN: FVT-ASOF-2-031..034, FVT-039, and the new `AsOfJoinSQLSuite`
tests. FVT-041 does cover the widening of the visible right-side columns. But
if this hidden-column widening were dropped, for example to plain `leftHidden
++ rightHidden`, no golden, `checkAnswer`, or dual-run comparison would change.
The single-pass plan would then declare `d.k` NOT NULL over the NULLs produced
for unmatched left rows.
A case that would catch it needs a hidden key that is NOT NULL below the
join and read above it:
```sql
SELECT l.lts, d.k
FROM using_left l LEFT ASOF JOIN (using_right r JOIN using_keys d USING (k))
MATCH_CONDITION (l.k <= d.k)
```
`l.k = 8` has no match, so `d.k` comes back NULL. The parent's `d.k`
reference is already widened through the scope's hidden output, so dual-run
would flag an unwidened wrapper attribute.
This extends an [existing
discussion](https://github.com/apache/spark/pull/59267#discussion_r4271050826).
**Verification:**
- **Behavior:** A LEFT ASOF JOIN whose MATCH_CONDITION uses a NOT NULL
hidden key, read by the parent, returns the expected rows, including NULL for
unmatched left rows, and its schema reports the column as nullable.
- **Regression:** Removing the join-type widening from the wrapper's
appended hidden columns makes this case fail under dual-run.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RetainsOriginalJoinOutput.scala:
##########
@@ -127,10 +127,30 @@ trait RetainsOriginalJoinOutput {
referencedAttributes = childReferencedAttributes
) =>
Project(scopes.current.output, join)
+ case asOfJoin: AsOfJoin
Review Comment:
Thanks, confirmed at 1c23257. AsOfJoinResolver now adds the restoring
Project itself, and buildUsingProject puts the USING Project on top of it.
FVT-ASOF-2-030 shows Project(using) <- Project [k, lts, k, rts] <- AsOfJoin,
which matches the fixed-point plan.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4257662381","thread_id":"inline:4257662381","verdict_sha256":"345736d3e98c123cab0949b6b6406ec4758f4c7d7f1ebcc67c52b399d935e133"}
-->
##########
sql/core/src/test/scala/org/apache/spark/sql/AsOfJoinSQLSuite.scala:
##########
@@ -157,6 +157,69 @@ class AsOfJoinSQLSuite extends QueryTest with
SharedSparkSession {
stop = 137)))
}
+ // Tests run both analyzers and compare their plans (dualRunWithLegacy), so
checkAnswer also
+ // covers the single-pass analyzer. LEFT JOIN leaves d.k NULL on some rows,
so a match on the
+ // visible k column gives a different answer.
+ test("MATCH_CONDITION accepts a hidden USING column of the right input") {
Review Comment:
Thanks, confirmed. FVT-ASOF-2-031 and 033 to 038 now cover a parent that
selects the hidden key, plus GROUP BY, WHERE, count(*), alias, and join-input
parents. Since the wrap moved into AsOfJoinResolver, the parent-specific path
this comment was about is also gone.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4257662392","thread_id":"inline:4257662392","verdict_sha256":"345736d3e98c123cab0949b6b6406ec4758f4c7d7f1ebcc67c52b399d935e133"}
-->
--
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]
