HyukjinKwon commented on code in PR #59267:
URL: https://github.com/apache/spark/pull/59267#discussion_r4286547006
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveAsOfJoin.scala:
##########
@@ -74,10 +74,12 @@ object ResolveAsOfJoin extends Rule[LogicalPlan] with
SQLConfHelper {
}
val resolvedJoin = (matchLeft, matchOp, matchRight) match {
case (Some(leftExpr), Some(operator), Some(rightExpr)) =>
+ val leftQualifiedAccessOnlyColumns =
left.metadataOutput.filter(_.qualifiedAccessOnly)
Review Comment:
**Nit (P3):** This side set reads `left.metadataOutput`. When the left input
is itself an `ASOF JOIN ... USING`, that input is the USING `Project` built
below (lines 100-106, unchanged), and its hidden-output tag carries only that
join's own `hiddenList`. The regular USING join also appends
`project.child.metadataOutput.filter(_.qualifiedAccessOnly)`
(`Analyzer.commonNaturalJoinProcessing`), but this tag drops the hidden columns
of nested USING joins. Single-pass keeps them: `buildUsingProject` derives its
hidden output from the overwritten scope, so the outer
`leftNameScope.hiddenOutput` still contains them.
As a result, the two analyzers now disagree on an outer `MATCH_CONDITION`
operand that names such a column:
```sql
SELECT l.lts, x.k
FROM (using_left l ASOF JOIN (using_right r LEFT JOIN using_keys d USING (k))
MATCH_CONDITION (l.lts >= r.rts) USING (k))
ASOF JOIN using_keys x MATCH_CONDITION (d.k <= x.k)
```
Fixed-point cannot resolve `d.k` and fails with UNRESOLVED_COLUMN.
Single-pass classifies it to the left side and resolves the query, so dual-run,
which is on by default in tests, raises `fixedPointFailedSinglePassSucceeded`.
Before this PR both analyzers rejected the query. The underlying gap predates
the PR (a parent `SELECT d.k` over the same ASOF USING join already diverges),
but the widened side sets make it reachable from `MATCH_CONDITION`. The default
fixed-point path behaves as before.
**Recommended change:** In the USING branch, tag the Project with
`hiddenList.map(_.markAsQualifiedAccessOnly()) ++
project.child.metadataOutput.filter(_.qualifiedAccessOnly)`, as the regular
USING join does. Then add an ASOF JOIN golden for the nested shape above.
**Verification:**
- **Regression:** Under dual-run, an outer ASOF JOIN whose MATCH_CONDITION
uses a hidden key nested under an inner ASOF JOIN ... USING resolves with
identical single-pass and fixed-point plans and returns the expected rows.
- **Compatibility:** Existing ASOF JOIN ... USING goldens, including
FVT-ASOF-2-030, keep their analyzed plans and results.
##########
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.
Review Comment:
**Nit (P3):** The wrap does not keep "the hidden columns" in general. It
restores the join's original output and appends only
`hiddenOutput.filter(_.qualifiedAccessOnly)`, and nothing at all when the join
is the top-level operator (FVT-ASOF-2-040's golden is `Project [k, lts, k,
rts]` without `d.k`). In this resolver, "hidden columns" usually means the
whole `NameScope.hiddenOutput`, so the current wording can mislead. Something
like:
```suggestion
* If the join references a hidden column, wraps it in a [[Project]] that
restores its original
* output, plus the qualified-access-only columns unless the join is the
top-level operator.
```
**Verification:**
- **Inspection:** The scaladoc names the original-output restoration and the
qualified-access-only, non-top-level condition, consistent with
withQualifiedAccessOnlyColumns.
##########
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:
Thanks! FVT-ASOF-2-041 is the LEFT ASOF case from this comment: `l.k = 8`
has no match and comes back with a NULL `rts`, so dual-run now compares the
LeftOuter wrapper. FVT-ASOF-2-042 also reads a NOT NULL hidden `d.k` above the
LEFT ASOF JOIN, which covers the widening of the wrapper's appended hidden
columns as well.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4271050826","thread_id":"inline:4271050826","verdict_sha256":"46f035ae3bb235bb60eee4b0415b9b24faaceac233a673342a2b00d79d5060bf"}
-->
--
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]
