vranes commented on code in PR #59834:
URL: https://github.com/apache/spark/pull/59834#discussion_r4410336735
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoinSuite.scala:
##########
@@ -277,39 +277,44 @@ class RewriteAsOfJoinSuite extends PlanTest {
comparePlans(rewritten, correctAnswer, checkAnalysis = false)
}
- test("no rewrite when the node requires the sort-merge as-of join operator")
{
- // A node flagged `requiresSortMergeAsOfJoin = true` (e.g. from a SQL
MATCH_CONDITION) is
- // left untouched so the sort-merge physical operator handles it instead.
- val query = AsOfJoin(left, right, left.output(0), right.output(0), None,
Inner,
- tolerance = None, allowExactMatches = true, direction =
AsOfJoinDirection("backward"))
- .copy(requiresSortMergeAsOfJoin = true)
- val analyzed = query.analyze
-
- // The flag must survive analysis, so it, not a dropped flag, is what
makes this a no-op.
- assert(analyzed.collectFirst { case a: AsOfJoin => a
}.exists(_.requiresSortMergeAsOfJoin),
- "expected the analyzed plan to still carry requiresSortMergeAsOfJoin =
true")
-
- val rewritten = RewriteAsOfJoin(analyzed)
-
- comparePlans(rewritten, analyzed)
- }
-
test("no rewrite when the sort-merge as-of join operator is enabled by
config") {
- // With the config on, the sort-merge operator is enabled globally, so
even a plain node
- // (flag off) is left intact rather than rewritten to a subquery.
+ // Analysis records the strategy, so later config changes do not affect
the rewrite.
withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "true") {
val query = AsOfJoin(left, right, left.output(0), right.output(0), None,
Inner,
tolerance = None, allowExactMatches = true, direction =
AsOfJoinDirection("backward"))
val analyzed = query.analyze
- comparePlans(RewriteAsOfJoin(analyzed), analyzed)
+ withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "false") {
+ comparePlans(RewriteAsOfJoin(analyzed), analyzed)
+ }
+ }
+ }
+
+ test("sort-merge normalization preserves the tolerance condition on repeated
application") {
+ withSQLConf(SQLConf.SORT_MERGE_AS_OF_JOIN_ENABLED.key -> "true") {
+ val analyzed = AsOfJoin(
+ left = left,
+ right = right,
+ leftAsOf = right.output(0),
+ rightAsOf = left.output(0),
+ condition = None,
+ joinType = Inner,
+ tolerance = Some(Literal(1)),
+ allowExactMatches = true,
+ direction = AsOfJoinDirection("backward")
+ ).analyze.asInstanceOf[AsOfJoin]
+
+ val normalized = RewriteAsOfJoin(analyzed).asInstanceOf[AsOfJoin]
+
+ assert(normalized.asOfCondition == analyzed.asOfCondition)
Review Comment:
Thanks, fixed. The unit test now asserts the expected tolerance condition.
Added backward/forward execution tests with reversed keys that check both
matches within the tolerance and rejection outside it, under both analyzers.
Also updated the PR description.
--
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]
