From 1d13af09cfc35ce34489979830f1c8581af4d069 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Luka=20Zdravi=C4=87?= Date: Thu, 10 Sep 2026 12:21:02 +0000 Subject: [PATCH] [SPARK-59409][SQL][TESTS] Expand RewriteAsOfJoin optimizer rule test coverage Refactor RewriteAsOfJoinSuite onto a shared `expectedRewrite` helper and add coverage for previously-untested parts of the RewriteAsOfJoin optimizer rule: - the missing nearest + allowExactMatches=false + no-tolerance combination (as-of condition NOT(left = right)); - a real LeftOuter case (the old "left outer" test built an Inner join); - the no-op guard, for both the requiresSortMergeAsOfJoin node flag and the sort-merge as-of config; - the attribute remapping exercised by an operator above the join. Test-only; no production changes. --- .../optimizer/RewriteAsOfJoinSuite.scala | 388 ++++++++++-------- 1 file changed, 219 insertions(+), 169 deletions(-) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoinSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoinSuite.scala index 09ed5f0f073c2..ca339558d7dfb 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoinSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteAsOfJoinSuite.scala @@ -19,24 +19,31 @@ package org.apache.spark.sql.catalyst.optimizer import org.apache.spark.sql.catalyst.dsl.expressions._ import org.apache.spark.sql.catalyst.dsl.plans._ -import org.apache.spark.sql.catalyst.expressions.{CreateStruct, GetStructField, If, OuterReference, ScalarSubquery} +import org.apache.spark.sql.catalyst.expressions.{CreateStruct, Expression, GetStructField, If, Literal, OuterReference, ScalarSubquery} import org.apache.spark.sql.catalyst.expressions.aggregate.MinBy -import org.apache.spark.sql.catalyst.plans.{AsOfJoinDirection, Inner, LeftOuter, PlanTest} -import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LocalRelation} +import org.apache.spark.sql.catalyst.plans.{AsOfJoinDirection, Inner, JoinType, LeftOuter, PlanTest} +import org.apache.spark.sql.catalyst.plans.logical.{AsOfJoin, LocalRelation, LogicalPlan} +import org.apache.spark.sql.internal.SQLConf class RewriteAsOfJoinSuite extends PlanTest { - test("simple") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) - val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, - tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("backward")) - - val rewritten = RewriteAsOfJoin(query.analyze) - - val filter = OuterReference(left.output(0)) >= right.output(0) + private val left = LocalRelation($"a".int, $"b".int, $"c".int) + private val right = LocalRelation($"a".int, $"b".int, $"d".int) + + // Builds the plan RewriteAsOfJoin is expected to produce. For each left row it runs a + // correlated scalar subquery that picks the nearest matching right row via + // MIN_BY(struct(right.*), orderExpression) over the right rows satisfying `filter`, then + // projects the right columns back out. INNER drops left rows with no match through + // `__right__ IS NOT NULL`; LEFT OUTER keeps them, so it omits that Filter. + // + // `filter` and `orderExpression` are the only parts that vary with + // (direction, tolerance, allowExactMatches): they mirror AsOfJoin.makeAsOfCond and + // makeOrderingExpr, so each test spells just those out and shares the rest of the shape here. + private def expectedRewrite( + filter: Expression, + orderExpression: Expression, + joinType: JoinType): LogicalPlan = { val rightStruct = CreateStruct(right.output) - val orderExpression = OuterReference(left.output(0)) - right.output(0) val nearestRight = MinBy(rightStruct, orderExpression) .toAggregateExpression().as("__nearest_right__") @@ -44,246 +51,289 @@ class RewriteAsOfJoinSuite extends PlanTest { left.output :+ ScalarSubquery( right.where(filter).groupBy()(nearestRight), left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + val withNullFilter = joinType match { + case LeftOuter => scalarSubquery + case _ => scalarSubquery.where(scalarSubquery.output.last.isNotNull) + } + withNullFilter.select(left.output ++ right.output.zipWithIndex.map { + case (attr, idx) => GetStructField(scalarSubquery.output.last, idx).as(attr.name) + }: _*) + } + + test("simple") { + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("backward")) + + val rewritten = RewriteAsOfJoin(query.analyze) + + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) >= right.output(0), + orderExpression = OuterReference(left.output(0)) - right.output(0), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("condition") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), Some(left.output(1) === right.output(1)), Inner, tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("backward")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = OuterReference(left.output(1)) === right.output(1) && - OuterReference(left.output(0)) >= right.output(0) - val rightStruct = CreateStruct(right.output) - val orderExpression = OuterReference(left.output(0)) - right.output(0) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") - - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + // The join condition is AND-ed in front of the as-of condition. + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(1)) === right.output(1) && + OuterReference(left.output(0)) >= right.output(0), + orderExpression = OuterReference(left.output(0)) - right.output(0), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("left outer") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) - val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, LeftOuter, tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("backward")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = OuterReference(left.output(0)) >= right.output(0) - val rightStruct = CreateStruct(right.output) - val orderExpression = OuterReference(left.output(0)) - right.output(0) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") - - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + // LEFT OUTER keeps left rows with no match, so the rewrite omits the `IS NOT NULL` filter + // that INNER applies. The condition and ordering are the same as the `simple` case. + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) >= right.output(0), + orderExpression = OuterReference(left.output(0)) - right.output(0), + joinType = LeftOuter) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("tolerance") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, tolerance = Some(1), allowExactMatches = true, direction = AsOfJoinDirection("backward")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = OuterReference(left.output(0)) >= right.output(0) && - right.output(0) >= OuterReference(left.output(0)) - 1 - val rightStruct = CreateStruct(right.output) - val orderExpression = OuterReference(left.output(0)) - right.output(0) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") - - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) >= right.output(0) && + right.output(0) >= OuterReference(left.output(0)) - 1, + orderExpression = OuterReference(left.output(0)) - right.output(0), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("allowExactMatches = false") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), None, LeftOuter, tolerance = None, allowExactMatches = false, direction = AsOfJoinDirection("backward")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = OuterReference(left.output(0)) > right.output(0) - val rightStruct = CreateStruct(right.output) - val orderExpression = OuterReference(left.output(0)) - right.output(0) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") - - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) > right.output(0), + orderExpression = OuterReference(left.output(0)) - right.output(0), + joinType = LeftOuter) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("tolerance & allowExactMatches = false") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, tolerance = Some(1), allowExactMatches = false, direction = AsOfJoinDirection("backward")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = OuterReference(left.output(0)) > right.output(0) && - right.output(0) > OuterReference(left.output(0)) - 1 - val rightStruct = CreateStruct(right.output) - val orderExpression = OuterReference(left.output(0)) - right.output(0) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") - - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) > right.output(0) && + right.output(0) > OuterReference(left.output(0)) - 1, + orderExpression = OuterReference(left.output(0)) - right.output(0), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("direction = forward") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("forward")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = OuterReference(left.output(0)) <= right.output(0) - val rightStruct = CreateStruct(right.output) - val orderExpression = right.output(0) - OuterReference(left.output(0)) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") + // Forward flips the comparison (`<=`) and the ordering distance (right - left). + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) <= right.output(0), + orderExpression = right.output(0) - OuterReference(left.output(0)), + joinType = Inner) - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + comparePlans(rewritten, correctAnswer, checkAnalysis = false) + } + + test("direction = forward & allowExactMatches = false") { + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = None, allowExactMatches = false, direction = AsOfJoinDirection("forward")) + + val rewritten = RewriteAsOfJoin(query.analyze) + + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) < right.output(0), + orderExpression = right.output(0) - OuterReference(left.output(0)), + joinType = Inner) + + comparePlans(rewritten, correctAnswer, checkAnalysis = false) + } + + test("tolerance & direction = forward") { + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = Some(1), allowExactMatches = true, direction = AsOfJoinDirection("forward")) + + val rewritten = RewriteAsOfJoin(query.analyze) + + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) <= right.output(0) && + right.output(0) <= OuterReference(left.output(0)) + 1, + orderExpression = right.output(0) - OuterReference(left.output(0)), + joinType = Inner) + + comparePlans(rewritten, correctAnswer, checkAnalysis = false) + } + + test("tolerance & allowExactMatches = false & direction = forward") { + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = Some(1), allowExactMatches = false, direction = AsOfJoinDirection("forward")) + + val rewritten = RewriteAsOfJoin(query.analyze) + + val correctAnswer = expectedRewrite( + filter = OuterReference(left.output(0)) < right.output(0) && + right.output(0) < OuterReference(left.output(0)) + 1, + orderExpression = right.output(0) - OuterReference(left.output(0)), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("direction = nearest") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("nearest")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = true - val rightStruct = CreateStruct(right.output) - val orderExpression = If(OuterReference(left.output(0)) > right.output(0), - OuterReference(left.output(0)) - right.output(0), - right.output(0) - OuterReference(left.output(0))) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") + // nearest + allowExactMatches with no tolerance places no constraint on the match, so the + // as-of condition collapses to the literal true. The ordering picks the smallest absolute + // distance in either direction. + val correctAnswer = expectedRewrite( + filter = Literal.TrueLiteral, + orderExpression = If(OuterReference(left.output(0)) > right.output(0), + OuterReference(left.output(0)) - right.output(0), + right.output(0) - OuterReference(left.output(0))), + joinType = Inner) - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + comparePlans(rewritten, correctAnswer, checkAnalysis = false) + } + + test("allowExactMatches = false & direction = nearest") { + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = None, allowExactMatches = false, direction = AsOfJoinDirection("nearest")) + + val rewritten = RewriteAsOfJoin(query.analyze) + + // nearest without tolerance and without exact matches: the only constraint is that the + // right key differs from the left key (AsOfJoin.makeAsOfCond `case (false, Nearest)`). + val correctAnswer = expectedRewrite( + filter = !(OuterReference(left.output(0)) === right.output(0)), + orderExpression = If(OuterReference(left.output(0)) > right.output(0), + OuterReference(left.output(0)) - right.output(0), + right.output(0) - OuterReference(left.output(0))), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } test("tolerance & allowExactMatches = false & direction = nearest") { - val left = LocalRelation($"a".int, $"b".int, $"c".int) - val right = LocalRelation($"a".int, $"b".int, $"d".int) val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, tolerance = Some(1), allowExactMatches = false, direction = AsOfJoinDirection("nearest")) val rewritten = RewriteAsOfJoin(query.analyze) - val filter = (!(OuterReference(left.output(0)) === right.output(0))) && - ((right.output(0) > OuterReference(left.output(0)) - 1) && - (right.output(0) < OuterReference(left.output(0)) + 1)) - val rightStruct = CreateStruct(right.output) - val orderExpression = If(OuterReference(left.output(0)) > right.output(0), - OuterReference(left.output(0)) - right.output(0), - right.output(0) - OuterReference(left.output(0))) - val nearestRight = MinBy(rightStruct, orderExpression) - .toAggregateExpression().as("__nearest_right__") + val correctAnswer = expectedRewrite( + filter = (!(OuterReference(left.output(0)) === right.output(0))) && + ((right.output(0) > OuterReference(left.output(0)) - 1) && + (right.output(0) < OuterReference(left.output(0)) + 1)), + orderExpression = If(OuterReference(left.output(0)) > right.output(0), + OuterReference(left.output(0)) - right.output(0), + right.output(0) - OuterReference(left.output(0))), + joinType = Inner) - val scalarSubquery = left.select( - left.output :+ ScalarSubquery( - right.where(filter).groupBy()(nearestRight), - left.output).as("__right__"): _*) - val correctAnswer = scalarSubquery - .where(scalarSubquery.output.last.isNotNull) - .select(left.output :+ - GetStructField(scalarSubquery.output.last, 0).as("a") :+ - GetStructField(scalarSubquery.output.last, 1).as("b") :+ - GetStructField(scalarSubquery.output.last, 2).as("d"): _*) + comparePlans(rewritten, correctAnswer, checkAnalysis = false) + } + + test("tolerance & direction = nearest") { + val query = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = Some(1), allowExactMatches = true, direction = AsOfJoinDirection("nearest")) + + val rewritten = RewriteAsOfJoin(query.analyze) + + // nearest + allowExactMatches + tolerance intentionally drops the `true` base condition and + // keeps only the two-sided tolerance band (AsOfJoin.makeAsOfCond `case (true, Nearest)`). + val correctAnswer = expectedRewrite( + filter = right.output(0) >= OuterReference(left.output(0)) - 1 && + right.output(0) <= OuterReference(left.output(0)) + 1, + orderExpression = If(OuterReference(left.output(0)) > right.output(0), + OuterReference(left.output(0)) - right.output(0), + right.output(0) - OuterReference(left.output(0))), + joinType = Inner) comparePlans(rewritten, correctAnswer, checkAnalysis = false) } + + test("no rewrite when the node requires the sort-merge as-of join operator") { + // The rule only fires when `!conf.useSortMergeAsOfJoinOperator(requiresSortMergeAsOfJoin)`. + // A node flagged `requiresSortMergeAsOfJoin = true` (e.g. from a SQL `MATCH_CONDITION` + // clause) must be 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 + + // Precondition: the flag survives analysis, so it is the guard -- not a dropped flag -- + // that 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") { + // The guard is `sortMergeAsOfJoinEnabled || requiresSortMergeAsOfJoin`. This is the other + // trigger: with the config on, the sort-merge operator is enabled globally, so even a plain + // node (flag off) is left intact for that operator rather than rewritten to a subquery. + 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) + } + } + + test("references above the join are remapped to the rewritten output") { + // Every other test roots the query at AsOfJoin, so the attribute remapping that + // `transformUpWithNewOutput` returns (the `project -> attrMapping` pair) never has an + // ancestor to fix up. Put a Project above the join that selects a right-side column: the + // rewrite gives that column a fresh exprId (a GetStructField alias), so the parent + // reference must be remapped onto it, otherwise the plan is left with a dangling reference. + val join = AsOfJoin(left, right, left.output(0), right.output(0), None, Inner, + tolerance = None, allowExactMatches = true, direction = AsOfJoinDirection("backward")) + val originalRightExprId = join.output.last.exprId + val query = join.select(join.output.last).analyze + + val rewritten = RewriteAsOfJoin(query) + + rewritten.foreach { node => + assert(node.missingInput.isEmpty, + s"${node.nodeName} has dangling references: ${node.missingInput}") + } + assert(!rewritten.references.exists(_.exprId == originalRightExprId), + "expected the parent projection to be remapped off the original AsOfJoin output") + } }