2010YOUY01 commented on code in PR #23829:
URL: https://github.com/apache/datafusion/pull/23829#discussion_r3910774720
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -4343,6 +4443,169 @@ pub struct Join {
pub null_aware: bool,
}
+/// The ordered comparison used by an [`AsOfJoin`].
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+pub struct AsOfMatch {
+ /// Expression evaluated against the left input.
+ pub left: Expr,
+ /// One of [`Operator::Lt`], [`Operator::LtEq`], [`Operator::Gt`], or
+ /// [`Operator::GtEq`].
+ pub op: Operator,
+ /// Expression evaluated against the right input.
+ pub right: Expr,
+}
+
+impl AsOfMatch {
+ /// Creates an ordered ASOF match condition.
+ pub fn new(left: Expr, op: Operator, right: Expr) -> Self {
+ Self { left, op, right }
+ }
+}
+
+impl Display for AsOfMatch {
+ fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+ write!(f, "{} {} {}", self.left, self.op, self.right)
+ }
+}
+
+/// Match each left row with at most one ordered row from the right input.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct AsOfJoin {
+ /// Left input. Every left row is preserved exactly once.
+ pub left: Arc<LogicalPlan>,
+ /// Right input.
+ pub right: Arc<LogicalPlan>,
+ /// Equality clauses expressed as pairs of left and right expressions.
+ pub on: Vec<(Expr, Expr)>,
+ /// Ordered match condition.
+ pub match_condition: Box<AsOfMatch>,
+ /// Whether equality keys came from `ON` or `USING`.
+ pub join_constraint: JoinConstraint,
+ /// Output schema.
+ pub schema: DFSchemaRef,
+}
+
+impl AsOfJoin {
+ /// Creates an ASOF join and validates its logical contract.
+ ///
+ /// This is the pre-coercion boundary. The physical ASOF constructor
repeats
Review Comment:
I find `Join::try_new()` (in the current file) don't do any validation,
maybe it's allowed to build invalid logical plan, and defer validation to the
physical plan construction?
After SQL integration we can try if we can delete the validation here to
simplify it, it's always better if we can consolidate all the related
validation to a single place.
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -1007,6 +1007,68 @@ impl LogicalPlanBuilder {
)
}
+ /// Apply a left-preserving ASOF join using equality expressions and one
+ /// ordered match condition.
+ pub fn asof_join(
+ self,
+ right: LogicalPlan,
+ on: Vec<(Expr, Expr)>,
+ match_condition: AsOfMatch,
Review Comment:
I think this is the user-facing API (see the example code in the `join_on`
comment above), so it should be easy to use.
Currently it asks callers to manually normalize the args; instead, we should
validate them internally.
Perhaps:
```rust
on: Expr,
match_condition: Expr,
```
For example, if the SQL input is `on t1.v1 < t2.v1`, which is not a
supported ON clause for an asof join:
- Existing design: we have to do partial validation in the SQL->LogicalPlan
binding
- Alternative: we can consolidate the validation in one place
But I think we can proceed as is and potentially change it in the SQL
integration PR. It would be obvious which design is better once we try to build
a plan from SQL. And we're likely to do that before the next release, so API
changes are fine.
##########
datafusion/core/src/physical_planner.rs:
##########
@@ -1790,6 +1791,51 @@ impl DefaultPhysicalPlanner {
join
}
}
+ LogicalPlan::AsOfJoin(join) => {
+ let [physical_left, physical_right] = children.two()?;
+ let join_on = join
+ .on
+ .iter()
+ .map(|(left, right)| {
+ Ok((
+ create_physical_expr(
+ left,
+ join.left.schema(),
+ execution_props,
+ planning_ctx,
+ )?,
+ create_physical_expr(
+ right,
+ join.right.schema(),
+ execution_props,
+ planning_ctx,
+ )?,
+ ))
+ })
+ .collect::<Result<join_utils::JoinOn>>()?;
+ let match_condition = AsOfMatchExpr::new(
+ create_physical_expr(
+ &join.match_condition.left,
+ join.left.schema(),
+ execution_props,
+ planning_ctx,
+ )?,
+ join.match_condition.op,
+ create_physical_expr(
+ &join.match_condition.right,
+ join.right.schema(),
+ execution_props,
+ planning_ctx,
+ )?,
+ );
+ Arc::new(AsOfJoinExec::try_new(
+ physical_left,
+ physical_right,
+ join_on,
+ match_condition,
+ None,
Review Comment:
Here is something I don't fully understand — it would be great if other
reviewers have the knowledge to double-check, or can point me to something that
would help me understand it more easily. (Just flagging where I don't have full
confidence; I don't think this is a blocker for this PR.)
Specifically, I don't follow the whole lifecycle of projection pushdown.
Let's say the downstream only requires a subset of the columns inside
`AsOfJoinExec`. I imagine the logical optimizer should try to keep the
projection list inside the `LogicalPlan`, so that we can directly build the
physical plan node with the required projection indices.
Here, when building the initial physical plan, the projection list is
`None`, and it seems to depend on a later physical optimizer pass to finish the
work.
Perhaps there is some room to simplify that process.
--
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]
