2010YOUY01 commented on code in PR #23829:
URL: https://github.com/apache/datafusion/pull/23829#discussion_r3888981910
##########
datafusion/optimizer/src/push_down_filter.rs:
##########
@@ -1115,6 +1115,35 @@ impl OptimizerRule for PushDownFilter {
result.map_data(|plan| Ok(with_filters(keep_predicates, plan)))
}
LogicalPlan::Join(join) => push_down_join(join,
Some(filter.predicate)),
+ LogicalPlan::AsOfJoin(mut join) => {
Review Comment:
Marker for filter pushdown, see comments for details.
##########
datafusion/expr/src/logical_plan/builder.rs:
##########
@@ -1776,6 +1838,15 @@ pub fn build_join_schema(
dfschema.with_functional_dependencies(func_dependencies)
}
+/// Creates the schema for a left-preserving ASOF join.
+///
+/// Both `ON` and `USING` preserve all qualified input fields. SQL wildcard
+/// expansion handles the unqualified `USING` key as a single column.
+pub fn build_asof_join_schema(left: &DFSchema, right: &DFSchema) ->
Result<DFSchema> {
+ build_join_schema(left, right, &JoinType::Left)?
+ .with_functional_dependencies(left.functional_dependencies().clone())
Review Comment:
Marker for functional dependency optimization, see comments for details.
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -4343,6 +4443,164 @@ 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.
+ pub fn try_new(
+ left: Arc<LogicalPlan>,
+ right: Arc<LogicalPlan>,
+ on: Vec<(Expr, Expr)>,
+ match_condition: AsOfMatch,
+ join_constraint: JoinConstraint,
+ ) -> Result<Self> {
Review Comment:
I think this validation should be quite similar to the builder for
`AsofJoinExec`, is there any way we can unify them?
It can be tricky due to the physical representation gap (`Expr` v.s.
`PhysicalExpr`), but at least we could add a comment to both places like
```
Please keep `AsOfJoin::try_new()` and `AsOfJoinExec::try_new()`'s validation
logic in sync
```
##########
datafusion/expr/src/logical_plan/plan.rs:
##########
@@ -4343,6 +4443,164 @@ 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.
+ pub fn try_new(
+ left: Arc<LogicalPlan>,
+ right: Arc<LogicalPlan>,
+ on: Vec<(Expr, Expr)>,
+ match_condition: AsOfMatch,
Review Comment:
nit and optional: would it be better to use `Expr` instead of `AsOfMatch`?
The reason is that the typed `AsOfMatch` can only do part of the validation.
It guarantees the expr has the form `expr1 comparator expr2`, but later we
still need to validate `expr1` and `expr2` further. I think keeping them in one
place might be better — the "scattered implementation" pattern is bad in most
cases.
This might just be personal taste though. I don't see it as a major issue,
happy to be convinced otherwise.
--
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]
