LukaZdravic commented on code in PR #59957:
URL: https://github.com/apache/spark/pull/59957#discussion_r4455885239
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/SortMergeAsOfJoinExec.scala:
##########
@@ -518,3 +605,54 @@ private[joins] class SortMergeAsOfJoinScanner(
if (lower.isNaN || upper.isNaN) 2 else if (lower.isInfinite ||
upper.isInfinite) 1 else 0
}
}
+
+private[execution] object SortMergeAsOfJoinScanner {
+
+ /**
+ * Whether the scan can read the right side once with a cursor, instead of
buffering each
+ * right-side group and scanning it again for every left row.
+ *
+ * This needs a backward or forward join with no residual condition, whose
as-of condition
+ * only compares the two sort keys. Both sides are sorted on exactly those
keys, so the right
+ * rows that pass the condition move forward as the left key grows. Anything
else, such as a
+ * tolerance, a tuple, or an operand cast that the sort key does not have,
keeps the buffer.
+ */
+ def supportsCursorScan(
+ direction: AsOfJoinDirection,
+ asOfCondition: Expression,
+ residualCondition: Option[Expression],
+ leftSortExprs: Seq[Expression],
+ rightSortExprs: Seq[Expression]): Boolean = {
+ // The operands as (lesser, greater).
+ val ordered = asOfCondition match {
+ case LessThan(lesser, greater) => Some((lesser, greater))
+ case LessThanOrEqual(lesser, greater) => Some((lesser, greater))
+ case GreaterThan(greater, lesser) => Some((lesser, greater))
+ case GreaterThanOrEqual(greater, lesser) => Some((lesser, greater))
+ case _ => None
+ }
+ // The operands as (left, right): backward needs left >= right, forward
left <= right.
+ val operands = (direction, ordered) match {
+ case (Backward, Some((lesser, greater))) => Some((greater, lesser))
+ case (Forward, Some((lesser, greater))) => Some((lesser, greater))
+ case _ => None
+ }
+ (leftSortExprs, rightSortExprs, operands) match {
+ case (Seq(leftKey), Seq(rightKey), Some((leftOperand, rightOperand))) =>
+ residualCondition.isEmpty &&
+ leftOperand.semanticEquals(leftKey) &&
+ rightOperand.semanticEquals(rightKey) &&
+ leftKey.dataType == rightKey.dataType &&
+ supportsCursorKey(leftKey.dataType)
+ case _ => false
+ }
+ }
+
+ /** Scalar key types only for now. STRUCT and ARRAY keys keep the buffered
scan. */
Review Comment:
Fixed comment!
--
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]
