You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/04 01:46:45 UTC
(datafusion) branch main updated: feat: add ASOF join logical semantics (#23829)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new e1942b1446 feat: add ASOF join logical semantics (#23829)
e1942b1446 is described below
commit e1942b1446e437515332ba8f09ce6b0cd6f46e75
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Sep 4 01:36:31 2026 +0000
feat: add ASOF join logical semantics (#23829)
## Which issue does this PR close?
- Part of #318.
- Umbrella PR: #23738.
- Depends on #23828 (merged).
## Rationale for this change
This is the logical-planning layer of the ASOF JOIN stack. It defines
the
logical contract and planner behavior separately from the SQL frontend
and
serialization formats.
#23828 is merged, so this PR's diff against `main` is the isolated
logical
layer. It no longer depends on the optional floating-point follow-up
#24375.
## What changes are included in this PR?
- Add `LogicalPlan::AsOfJoin`, `AsOfJoin`, and `AsOfMatch`.
- Validate deterministic expressions, input ownership, supported match
operators, equality-key types, and USING constraints.
- Add `LogicalPlanBuilder` entry points and schema construction that
preserves
both qualified `USING` keys while exposing one unqualified wildcard key.
- Integrate ASOF joins with tree transforms, display, type coercion,
projection
pruning, row bounds, and physical planning.
- Plan logical ASOF joins to the broadcast-based `AsOfJoinExec` from
#23828.
- Fail closed at proto, SQL unparser, and Substrait boundaries until
their
owning stack layers add explicit support.
- Defer ASOF-specific functional-dependency refinement to #24799 and
filter
pushdown to #24801 so each optimization can be reviewed independently.
## Are these changes tested?
Yes:
- `cargo fmt --all`
- `cargo clippy --all-targets --all-features -- -D warnings`
- `cargo test -p datafusion-expr min_rows_of_joins --all-features`
- `cargo test -p datafusion-substrait
asof_join_fails_closed_until_substrait_has_an_extension --all-features`
- The extended workspace test command from the contributor guide
## Are there any user-facing changes?
This adds logical-plan and builder APIs for ASOF joins. SQL syntax,
DataFrame
APIs, and plan serialization are intentionally left to dependent stack
PRs.
Floating equality keys remain rejected by the merged physical operator
unless
the independent follow-up #24375 is also included.
As with any new public `LogicalPlan` variant, downstream exhaustive
matches must
add an arm. The variant is appended so existing variants retain their
`PartialOrd` ordering; maintainers should still treat the enum addition
as a
Rust source-compatibility break.
This PR can be reviewed independently now that #23828 has merged. The
optimization follow-ups #24799 and #24801 are not required by the core
ASOF
stack.
---
datafusion/core/src/physical_planner.rs | 49 +++-
datafusion/expr/src/logical_plan/builder.rs | 78 ++++-
datafusion/expr/src/logical_plan/display.rs | 23 +-
datafusion/expr/src/logical_plan/mod.rs | 16 +-
datafusion/expr/src/logical_plan/plan.rs | 320 ++++++++++++++++++++-
datafusion/expr/src/logical_plan/tree_node.rs | 58 +++-
datafusion/optimizer/src/analyzer/type_coercion.rs | 39 ++-
.../optimizer/src/common_subexpr_eliminate.rs | 1 +
.../optimizer/src/optimize_projections/mod.rs | 20 ++
datafusion/optimizer/src/optimizer.rs | 5 +
datafusion/physical-plan/src/joins/asof_join.rs | 4 +
datafusion/proto/src/logical_plan/mod.rs | 3 +
datafusion/sql/src/unparser/plan.rs | 1 +
.../substrait/src/logical_plan/producer/rel/mod.rs | 3 +
datafusion/substrait/tests/cases/serialize.rs | 27 +-
15 files changed, 608 insertions(+), 39 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index 9dde89fb1c..882956a63c 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -41,7 +41,8 @@ use crate::physical_plan::explain::ExplainExec;
use crate::physical_plan::filter::FilterExecBuilder;
use crate::physical_plan::joins::utils as join_utils;
use crate::physical_plan::joins::{
- CrossJoinExec, HashJoinExec, NestedLoopJoinExec, PartitionMode, SortMergeJoinExec,
+ AsOfJoinExec, AsOfMatchExpr, CrossJoinExec, HashJoinExec, NestedLoopJoinExec,
+ PartitionMode, SortMergeJoinExec,
};
use crate::physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
use crate::physical_plan::projection::{ProjectionExec, ProjectionExpr};
@@ -1787,6 +1788,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,
+ )?)
+ }
LogicalPlan::RecursiveQuery(RecursiveQuery {
name,
is_distinct,
@@ -2288,6 +2334,7 @@ fn extract_dml_filters(
| LogicalPlan::Sort(_)
| LogicalPlan::Union(_)
| LogicalPlan::Join(_)
+ | LogicalPlan::AsOfJoin(_)
| LogicalPlan::Repartition(_)
| LogicalPlan::Aggregate(_)
| LogicalPlan::Window(_)
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index ceb2c124e2..a6a254ea25 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -31,10 +31,10 @@ use crate::expr_rewriter::{
rewrite_sort_cols_by_aggs,
};
use crate::logical_plan::{
- Aggregate, Analyze, Distinct, DistinctOn, EmptyRelation, Explain, Filter, Join,
- JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning, PlanType, Prepare,
- Projection, Repartition, Sort, SubqueryAlias, TableScanBuilder, Union, Unnest,
- Values, Window,
+ Aggregate, Analyze, AsOfJoin, AsOfMatch, Distinct, DistinctOn, EmptyRelation,
+ Explain, Filter, Join, JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning,
+ PlanType, Prepare, Projection, Repartition, Sort, SubqueryAlias, TableScanBuilder,
+ Union, Unnest, Values, Window,
};
use crate::select_expr::SelectExpr;
use crate::utils::{
@@ -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,
+ ) -> Result<Self> {
+ self.asof_join_with_constraint(right, on, match_condition, JoinConstraint::On)
+ }
+
+ /// Apply a left-preserving ASOF join using `USING` equality keys.
+ pub fn asof_join_using(
+ self,
+ right: LogicalPlan,
+ using_keys: Vec<Column>,
+ match_condition: AsOfMatch,
+ ) -> Result<Self> {
+ let on = using_keys
+ .into_iter()
+ .map(|key| {
+ let left = Self::normalize(&self.plan, key.clone())?;
+ let right = Self::normalize(&right, key)?;
+ Ok((Expr::Column(left), Expr::Column(right)))
+ })
+ .collect::<Result<_>>()?;
+ self.asof_join_with_constraint(right, on, match_condition, JoinConstraint::Using)
+ }
+
+ fn asof_join_with_constraint(
+ self,
+ right: LogicalPlan,
+ on: Vec<(Expr, Expr)>,
+ match_condition: AsOfMatch,
+ join_constraint: JoinConstraint,
+ ) -> Result<Self> {
+ let normalize = |expr, schema: &DFSchema| {
+ normalize_col_with_schemas_and_ambiguity_check(expr, &[&[schema]], &[])
+ };
+ let on = on
+ .into_iter()
+ .map(|(left, right_expr)| {
+ Ok((
+ normalize(left, self.plan.schema())?,
+ normalize(right_expr, right.schema())?,
+ ))
+ })
+ .collect::<Result<_>>()?;
+ let match_condition = AsOfMatch {
+ left: normalize(match_condition.left, self.plan.schema())?,
+ op: match_condition.op,
+ right: normalize(match_condition.right, right.schema())?,
+ };
+ Ok(Self::new(LogicalPlan::AsOfJoin(AsOfJoin::try_new(
+ self.plan,
+ Arc::new(right),
+ on,
+ match_condition,
+ join_constraint,
+ )?)))
+ }
+
pub(crate) fn normalize(plan: &LogicalPlan, column: Column) -> Result<Column> {
if column.relation.is_some() {
// column is already normalized
@@ -1778,6 +1840,14 @@ 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)
+}
+
/// (Re)qualify the sides of a join if needed, i.e. if the columns from one side would otherwise
/// conflict with the columns from the other.
/// This is especially useful for queries that come as Substrait, since Substrait doesn't currently allow specifying
diff --git a/datafusion/expr/src/logical_plan/display.rs b/datafusion/expr/src/logical_plan/display.rs
index 4afbb56702..3009253f53 100644
--- a/datafusion/expr/src/logical_plan/display.rs
+++ b/datafusion/expr/src/logical_plan/display.rs
@@ -21,10 +21,10 @@ use std::collections::HashMap;
use std::fmt;
use crate::{
- Aggregate, DescribeTable, Distinct, DistinctOn, DmlStatement, Expr, Filter, Join,
- Limit, LogicalPlan, Partitioning, Projection, RecursiveQuery, Repartition, Sort,
- Subquery, SubqueryAlias, TableProviderFilterPushDown, TableScan, Unnest, Values,
- Window, expr_vec_fmt,
+ Aggregate, AsOfJoin, DescribeTable, Distinct, DistinctOn, DmlStatement, Expr, Filter,
+ Join, Limit, LogicalPlan, Partitioning, Projection, RecursiveQuery, Repartition,
+ Sort, Subquery, SubqueryAlias, TableProviderFilterPushDown, TableScan, Unnest,
+ Values, Window, expr_vec_fmt,
};
use crate::dml::CopyTo;
@@ -493,6 +493,21 @@ impl<'a, 'b> PgJsonVisitor<'a, 'b> {
"Filter": format!("{}", filter_expr)
})
}
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ on,
+ match_condition,
+ join_constraint,
+ ..
+ }) => {
+ let join_expr: Vec<String> =
+ on.iter().map(|(l, r)| format!("{l} = {r}")).collect();
+ json!({
+ "Node Type": "AsOf Join",
+ "Join Constraint": format!("{join_constraint:?}"),
+ "Join Keys": join_expr.join(", "),
+ "Match Condition": match_condition.to_string(),
+ })
+ }
LogicalPlan::Repartition(Repartition {
partitioning_scheme,
..
diff --git a/datafusion/expr/src/logical_plan/mod.rs b/datafusion/expr/src/logical_plan/mod.rs
index 4766c3f333..98113d12c1 100644
--- a/datafusion/expr/src/logical_plan/mod.rs
+++ b/datafusion/expr/src/logical_plan/mod.rs
@@ -28,8 +28,8 @@ pub mod tree_node;
pub use builder::{
LogicalPlanBuilder, LogicalPlanBuilderOptions, LogicalTableSource, UNNAMED_TABLE,
- build_join_schema, requalify_sides_if_needed, table_scan, union,
- wrap_projection_for_join_if_necessary,
+ build_asof_join_schema, build_join_schema, requalify_sides_if_needed, table_scan,
+ union, wrap_projection_for_join_if_necessary,
};
pub use ddl::{
CreateCatalog, CreateCatalogSchema, CreateExternalTable, CreateFunction,
@@ -41,12 +41,12 @@ pub use dml::{
WriteOp,
};
pub use plan::{
- Aggregate, Analyze, ColumnUnnestList, DescribeTable, Distinct, DistinctOn,
- EmptyRelation, Explain, ExplainOption, Extension, FetchType, Filter, Join,
- JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning, PlanType, Projection,
- RangePartitioning, RecursiveQuery, Repartition, SkipType, Sort, StringifiedPlan,
- Subquery, SubqueryAlias, TableScan, TableScanBuilder, ToStringifiedPlan, Union,
- Unnest, Values, Window, projection_schema,
+ Aggregate, Analyze, AsOfJoin, AsOfMatch, ColumnUnnestList, DescribeTable, Distinct,
+ DistinctOn, EmptyRelation, Explain, ExplainOption, Extension, FetchType, Filter,
+ Join, JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning, PlanType,
+ Projection, RangePartitioning, RecursiveQuery, Repartition, SkipType, Sort,
+ StringifiedPlan, Subquery, SubqueryAlias, TableScan, TableScanBuilder,
+ ToStringifiedPlan, Union, Unnest, Values, Window, projection_schema,
};
pub use statement::{
Deallocate, Execute, Prepare, ResetVariable, SetVariable, Statement,
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 4e039e3e77..74b5c42353 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -41,14 +41,15 @@ use crate::logical_plan::display::{GraphvizVisitor, IndentVisitor};
use crate::logical_plan::extension::UserDefinedLogicalNode;
use crate::logical_plan::{DmlStatement, Statement, WriteOp};
use crate::utils::{
- check_aggregate_and_window_nesting, enumerate_grouping_sets, exprlist_to_fields,
- find_out_reference_exprs, grouping_set_expr_count, grouping_set_to_exprlist,
- merge_schema, split_conjunction,
+ check_aggregate_and_window_nesting, enumerate_grouping_sets, expr_to_columns,
+ exprlist_to_fields, find_out_reference_exprs, grouping_set_expr_count,
+ grouping_set_to_exprlist, merge_schema, split_conjunction,
};
use crate::{
BinaryExpr, CreateMemoryTable, CreateView, Execute, Expr, ExprSchemable, GroupingSet,
LogicalPlanBuilder, Operator, Prepare, TableProviderFilterPushDown, TableSource,
- WindowFunctionDefinition, build_join_schema, expr_vec_fmt, requalify_sides_if_needed,
+ WindowFunctionDefinition, build_asof_join_schema, build_join_schema, expr_vec_fmt,
+ requalify_sides_if_needed,
};
use crate::statistics::StatisticsRequest;
@@ -295,6 +296,9 @@ pub enum LogicalPlan {
Unnest(Unnest),
/// A variadic query (e.g. "Recursive CTEs")
RecursiveQuery(RecursiveQuery),
+ /// Match each left row with at most one ordered row from the right input.
+ /// This is used to implement SQL `ASOF JOIN`.
+ AsOfJoin(AsOfJoin),
}
impl Default for LogicalPlan {
@@ -342,6 +346,7 @@ impl LogicalPlan {
LogicalPlan::Aggregate(Aggregate { schema, .. }) => schema,
LogicalPlan::Sort(Sort { input, .. }) => input.schema(),
LogicalPlan::Join(Join { schema, .. }) => schema,
+ LogicalPlan::AsOfJoin(AsOfJoin { schema, .. }) => schema,
LogicalPlan::Repartition(Repartition { input, .. }) => input.schema(),
LogicalPlan::Limit(Limit { input, .. }) => input.schema(),
LogicalPlan::Statement(statement) => statement.schema(),
@@ -370,7 +375,8 @@ impl LogicalPlan {
| LogicalPlan::Projection(_)
| LogicalPlan::Aggregate(_)
| LogicalPlan::Unnest(_)
- | LogicalPlan::Join(_) => self
+ | LogicalPlan::Join(_)
+ | LogicalPlan::AsOfJoin(_) => self
.inputs()
.iter()
.map(|input| input.schema().as_ref())
@@ -460,6 +466,9 @@ impl LogicalPlan {
LogicalPlan::Aggregate(Aggregate { input, .. }) => vec![input],
LogicalPlan::Sort(Sort { input, .. }) => vec![input],
LogicalPlan::Join(Join { left, right, .. }) => vec![left, right],
+ LogicalPlan::AsOfJoin(AsOfJoin { left, right, .. }) => {
+ vec![left, right]
+ }
LogicalPlan::Limit(Limit { input, .. }) => vec![input],
LogicalPlan::Subquery(Subquery { subquery, .. }) => vec![subquery],
LogicalPlan::SubqueryAlias(SubqueryAlias { input, .. }) => vec![input],
@@ -495,12 +504,20 @@ impl LogicalPlan {
let mut using_columns: Vec<HashSet<Column>> = vec![];
self.apply_with_subqueries(|plan| {
- if let LogicalPlan::Join(Join {
- join_constraint: JoinConstraint::Using,
- on,
- ..
- }) = plan
- {
+ let on = match plan {
+ LogicalPlan::Join(Join {
+ join_constraint: JoinConstraint::Using,
+ on,
+ ..
+ })
+ | LogicalPlan::AsOfJoin(AsOfJoin {
+ join_constraint: JoinConstraint::Using,
+ on,
+ ..
+ }) => Some(on),
+ _ => None,
+ };
+ if let Some(on) = on {
// The join keys in using-join must be columns.
let columns =
on.iter().try_fold(HashSet::new(), |mut accumu, (l, r)| {
@@ -568,6 +585,7 @@ impl LogicalPlan {
right.head_output_expr()
}
},
+ LogicalPlan::AsOfJoin(AsOfJoin { left, .. }) => left.head_output_expr(),
LogicalPlan::RecursiveQuery(RecursiveQuery { static_term, .. }) => {
static_term.head_output_expr()
}
@@ -691,6 +709,26 @@ impl LogicalPlan {
null_aware,
}))
}
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ left,
+ right,
+ on,
+ match_condition,
+ join_constraint,
+ schema: _,
+ }) => Ok(LogicalPlan::AsOfJoin(AsOfJoin::try_new(
+ left,
+ right,
+ on.into_iter()
+ .map(|(left, right)| (left.unalias(), right.unalias()))
+ .collect(),
+ AsOfMatch {
+ left: match_condition.left.unalias(),
+ op: match_condition.op,
+ right: match_condition.right.unalias(),
+ },
+ join_constraint,
+ )?)),
LogicalPlan::Subquery(_) => Ok(self),
LogicalPlan::SubqueryAlias(SubqueryAlias {
input,
@@ -994,6 +1032,45 @@ impl LogicalPlan {
null_aware: *null_aware,
}))
}
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ on,
+ match_condition,
+ join_constraint,
+ ..
+ }) => {
+ let (left, right) = self.only_two_inputs(inputs)?;
+ let expected = on.len() * 2 + 2;
+ assert_eq_or_internal_err!(
+ expected,
+ expr.len(),
+ "Invalid number of new ASOF join expressions: expected {}, got {}",
+ expected,
+ expr.len()
+ );
+
+ let mut iter = expr.into_iter();
+ let mut new_on = Vec::with_capacity(on.len());
+ for _ in 0..on.len() {
+ let left = iter.next().expect("expression count checked").unalias();
+ let right = iter.next().expect("expression count checked").unalias();
+ new_on.push((left, right));
+ }
+ let match_left = iter.next().expect("expression count checked").unalias();
+ let match_right =
+ iter.next().expect("expression count checked").unalias();
+
+ Ok(LogicalPlan::AsOfJoin(AsOfJoin::try_new(
+ Arc::new(left),
+ Arc::new(right),
+ new_on,
+ AsOfMatch {
+ left: match_left,
+ op: match_condition.op,
+ right: match_right,
+ },
+ *join_constraint,
+ )?))
+ }
LogicalPlan::Subquery(Subquery {
outer_ref_columns,
spans,
@@ -1419,6 +1496,7 @@ impl LogicalPlan {
right.max_rows()
}
},
+ LogicalPlan::AsOfJoin(AsOfJoin { left, .. }) => left.max_rows(),
LogicalPlan::Repartition(Repartition { input, .. }) => input.max_rows(),
LogicalPlan::Union(Union { inputs, .. }) => {
inputs.iter().try_fold(0usize, |mut acc, plan| {
@@ -1494,6 +1572,7 @@ impl LogicalPlan {
| JoinType::LeftAnti
| JoinType::RightAnti => 0,
},
+ LogicalPlan::AsOfJoin(AsOfJoin { left, .. }) => left.min_rows(),
LogicalPlan::Union(Union { inputs, .. }) => inputs
.iter()
.fold(0, |rows, input| rows.saturating_add(input.min_rows())),
@@ -1547,6 +1626,7 @@ impl LogicalPlan {
LogicalPlan::Window(_) => Ok(None),
LogicalPlan::Aggregate(_) => Ok(None),
LogicalPlan::Join(_) => Ok(None),
+ LogicalPlan::AsOfJoin(_) => Ok(None),
LogicalPlan::Repartition(_) => Ok(None),
LogicalPlan::Union(_) => Ok(None),
LogicalPlan::EmptyRelation(_) => Ok(None),
@@ -1585,6 +1665,7 @@ impl LogicalPlan {
LogicalPlan::Window(_) => Ok(None),
LogicalPlan::Aggregate(_) => Ok(None),
LogicalPlan::Join(_) => Ok(None),
+ LogicalPlan::AsOfJoin(_) => Ok(None),
LogicalPlan::Repartition(_) => Ok(None),
LogicalPlan::Union(_) => Ok(None),
LogicalPlan::EmptyRelation(_) => Ok(None),
@@ -2213,6 +2294,25 @@ impl LogicalPlan {
}
}
}
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ on,
+ match_condition,
+ join_constraint,
+ ..
+ }) => {
+ let equality = on
+ .iter()
+ .map(|(left, right)| format!("{left} = {right}"))
+ .join(", ");
+ write!(
+ f,
+ "AsOf Join: match=[{match_condition}], constraint={join_constraint:?}"
+ )?;
+ if !equality.is_empty() {
+ write!(f, ", on=[{equality}]")?;
+ }
+ Ok(())
+ }
LogicalPlan::Repartition(Repartition {
partitioning_scheme,
..
@@ -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
+ /// the shared operator, side-ownership, and determinism checks for direct
+ /// physical-plan callers and adds execution-only constraints. Keep the
+ /// shared checks aligned across both entry points.
+ pub fn try_new(
+ left: Arc<LogicalPlan>,
+ right: Arc<LogicalPlan>,
+ on: Vec<(Expr, Expr)>,
+ match_condition: AsOfMatch,
+ join_constraint: JoinConstraint,
+ ) -> Result<Self> {
+ if !matches!(
+ match_condition.op,
+ Operator::Lt | Operator::LtEq | Operator::Gt | Operator::GtEq
+ ) {
+ return plan_err!(
+ "ASOF MATCH_CONDITION requires <, <=, >, or >=, found {}",
+ match_condition.op
+ );
+ }
+
+ Self::validate_side(&match_condition.left, left.schema(), "left match")?;
+ Self::validate_side(&match_condition.right, right.schema(), "right match")?;
+ if match_condition.left.is_volatile() || match_condition.right.is_volatile() {
+ return plan_err!("ASOF MATCH_CONDITION must be deterministic");
+ }
+
+ let left_type = match_condition.left.get_type(left.schema())?;
+ let right_type = match_condition.right.get_type(right.schema())?;
+ if crate::type_coercion::binary::comparison_coercion(&left_type, &right_type)
+ .is_none()
+ {
+ return plan_err!(
+ "ASOF match expressions have incompatible types {left_type} and {right_type}"
+ );
+ }
+
+ for (left_expr, right_expr) in &on {
+ Self::validate_side(left_expr, left.schema(), "left equality")?;
+ Self::validate_side(right_expr, right.schema(), "right equality")?;
+ if left_expr.is_volatile() || right_expr.is_volatile() {
+ return plan_err!("ASOF equality expressions must be deterministic");
+ }
+ let left_type = left_expr.get_type(left.schema())?;
+ let right_type = right_expr.get_type(right.schema())?;
+ let Some(common_type) = crate::type_coercion::binary::comparison_coercion(
+ &left_type,
+ &right_type,
+ ) else {
+ return plan_err!(
+ "ASOF equality expressions have incompatible types {left_type} and {right_type}"
+ );
+ };
+ if !crate::utils::can_hash(&common_type) {
+ return plan_err!(
+ "ASOF equality expressions have unsupported hash type {common_type}"
+ );
+ }
+ }
+
+ if join_constraint == JoinConstraint::Using
+ && on.iter().any(|(left, right)| {
+ left.get_as_join_column().is_none()
+ || right.get_as_join_column().is_none()
+ })
+ {
+ return plan_err!("ASOF USING keys must be columns");
+ }
+
+ let schema = build_asof_join_schema(left.schema(), right.schema())?;
+ Ok(Self {
+ left,
+ right,
+ on,
+ match_condition: Box::new(match_condition),
+ join_constraint,
+ schema: Arc::new(schema),
+ })
+ }
+
+ fn validate_side(expr: &Expr, schema: &DFSchema, name: &str) -> Result<()> {
+ let mut columns = HashSet::new();
+ expr_to_columns(expr, &mut columns)?;
+ if columns.is_empty() {
+ return plan_err!("ASOF {name} expression must reference its input");
+ }
+ if let Some(column) = columns
+ .iter()
+ .find(|column| !schema.is_column_from_schema(column))
+ {
+ return plan_err!(
+ "ASOF {name} expression references column {column} outside its input"
+ );
+ }
+ Ok(())
+ }
+}
+
+impl PartialOrd for AsOfJoin {
+ fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+ (
+ &self.left,
+ &self.right,
+ &self.on,
+ &self.match_condition,
+ &self.join_constraint,
+ )
+ .partial_cmp(&(
+ &other.left,
+ &other.right,
+ &other.on,
+ &other.match_condition,
+ &other.join_constraint,
+ ))
+ .filter(|cmp| *cmp != Ordering::Equal || self == other)
+ }
+}
+
impl Join {
/// Creates a new Join operator with automatically computed schema.
///
@@ -5915,6 +6178,15 @@ mod tests {
.build()?;
assert_eq!(cross_join.min_rows(), 2);
+ let asof_join = LogicalPlanBuilder::from(two_rows.clone())
+ .asof_join(
+ one_row.clone(),
+ vec![],
+ AsOfMatch::new(col("l.column1"), Operator::GtEq, col("r.column1")),
+ )?
+ .build()?;
+ assert_eq!(asof_join.min_rows(), 2);
+
for (join_type, expected_min_rows) in [
// An inner join with a join condition may filter out every row,
// while outer joins preserve the rows of the outer side(s).
@@ -6840,6 +7112,32 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_asof_using_preserves_qualified_keys() -> Result<()> {
+ let schema = Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("ts", DataType::Int64, false),
+ ]);
+ let left = Arc::new(table_scan(Some("t1"), &schema, None)?.build()?);
+ let right = Arc::new(table_scan(Some("t2"), &schema, None)?.build()?);
+ let join = AsOfJoin::try_new(
+ left,
+ right,
+ vec![(col("t1.id"), col("t2.id"))],
+ AsOfMatch::new(col("t1.ts"), Operator::GtEq, col("t2.ts")),
+ JoinConstraint::Using,
+ )?;
+
+ assert_eq!(join.schema.fields().len(), 4);
+ assert_eq!(
+ join.schema
+ .index_of_column(&Column::from_qualified_name("t2.id"))?,
+ 2
+ );
+ assert!(join.schema.field(2).is_nullable());
+ Ok(())
+ }
+
#[test]
fn test_join_try_new_schema_validation() -> Result<()> {
let left_schema = Schema::new(vec![
diff --git a/datafusion/expr/src/logical_plan/tree_node.rs b/datafusion/expr/src/logical_plan/tree_node.rs
index c4c1d743b5..ee43666736 100644
--- a/datafusion/expr/src/logical_plan/tree_node.rs
+++ b/datafusion/expr/src/logical_plan/tree_node.rs
@@ -41,11 +41,12 @@ use std::sync::Arc;
use crate::logical_plan::plan::RangePartitioning;
use crate::{
- Aggregate, Analyze, CreateMemoryTable, CreateView, DdlStatement, Distinct,
- DistinctOn, DmlStatement, Execute, Explain, Expr, Extension, Filter, Join, Limit,
- LogicalPlan, Partitioning, Prepare, Projection, RecursiveQuery, Repartition, Sort,
- Statement, Subquery, SubqueryAlias, TableScan, Union, Unnest, UserDefinedLogicalNode,
- Values, Window, WriteOp, builder::unnest_with_options, dml::CopyTo,
+ Aggregate, Analyze, AsOfJoin, AsOfMatch, CreateMemoryTable, CreateView, DdlStatement,
+ Distinct, DistinctOn, DmlStatement, Execute, Explain, Expr, Extension, Filter, Join,
+ Limit, LogicalPlan, Partitioning, Prepare, Projection, RecursiveQuery, Repartition,
+ Sort, Statement, Subquery, SubqueryAlias, TableScan, Union, Unnest,
+ UserDefinedLogicalNode, Values, Window, WriteOp, builder::unnest_with_options,
+ dml::CopyTo,
};
use datafusion_common::tree_node::TreeNodeRefContainer;
@@ -150,6 +151,23 @@ impl TreeNode for LogicalPlan {
null_aware,
})
}),
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ left,
+ right,
+ on,
+ match_condition,
+ join_constraint,
+ schema,
+ }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ left,
+ right,
+ on,
+ match_condition,
+ join_constraint,
+ schema,
+ })
+ }),
LogicalPlan::Limit(Limit { skip, fetch, input }) => input
.map_elements(f)?
.update_data(|input| LogicalPlan::Limit(Limit { skip, fetch, input })),
@@ -447,6 +465,13 @@ impl LogicalPlan {
LogicalPlan::Join(Join { on, filter, .. }) => {
(on, filter).apply_ref_elements(f)
}
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ on,
+ match_condition,
+ ..
+ }) => on.apply_elements(&mut f)?.visit_sibling(|| {
+ (&match_condition.left, &match_condition.right).apply_ref_elements(&mut f)
+ }),
LogicalPlan::Sort(Sort { expr, .. }) => expr.apply_elements(f),
LogicalPlan::Extension(extension) => {
// would be nice to avoid this copy -- maybe can
@@ -614,6 +639,29 @@ impl LogicalPlan {
null_aware,
})
}),
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ left,
+ right,
+ on,
+ match_condition,
+ join_constraint,
+ schema,
+ }) => (on, (match_condition.left, match_condition.right))
+ .map_elements(f)?
+ .update_data(|(on, (left_match, right_match))| {
+ LogicalPlan::AsOfJoin(AsOfJoin {
+ left,
+ right,
+ on,
+ match_condition: Box::new(AsOfMatch {
+ left: left_match,
+ op: match_condition.op,
+ right: right_match,
+ }),
+ join_constraint,
+ schema,
+ })
+ }),
LogicalPlan::Sort(Sort { expr, input, fetch }) => expr
.map_elements(f)?
.update_data(|expr| LogicalPlan::Sort(Sort { expr, input, fetch })),
diff --git a/datafusion/optimizer/src/analyzer/type_coercion.rs b/datafusion/optimizer/src/analyzer/type_coercion.rs
index 13b9541b6a..07d3173d63 100644
--- a/datafusion/optimizer/src/analyzer/type_coercion.rs
+++ b/datafusion/optimizer/src/analyzer/type_coercion.rs
@@ -57,10 +57,10 @@ use datafusion_expr::type_coercion::{
};
use datafusion_expr::utils::merge_schema;
use datafusion_expr::{
- Cast, DmlStatement, Expr, ExprSchemable, Join, Limit, LogicalPlan, Operator,
- Projection, Union, ValueOrLambda, WindowFrame, WindowFrameBound, WindowFrameUnits,
- WriteOp, is_false, is_not_false, is_not_true, is_not_unknown, is_true, is_unknown,
- lit, not,
+ AsOfJoin, AsOfMatch, Cast, DmlStatement, Expr, ExprSchemable, Join, Limit,
+ LogicalPlan, Operator, Projection, Union, ValueOrLambda, WindowFrame,
+ WindowFrameBound, WindowFrameUnits, WriteOp, is_false, is_not_false, is_not_true,
+ is_not_unknown, is_true, is_unknown, lit, not,
};
/// Performs type coercion by determining the schema
@@ -191,6 +191,7 @@ impl<'a> TypeCoercionRewriter<'a> {
pub fn coerce_plan(&mut self, plan: LogicalPlan) -> Result<LogicalPlan> {
match plan {
LogicalPlan::Join(join) => self.coerce_join(join),
+ LogicalPlan::AsOfJoin(join) => self.coerce_asof_join(join),
LogicalPlan::Union(union) => Self::coerce_union(union),
LogicalPlan::Limit(limit) => Self::coerce_limit(limit),
LogicalPlan::Dml(dml) => self.coerce_dml(dml),
@@ -284,6 +285,36 @@ impl<'a> TypeCoercionRewriter<'a> {
Ok(LogicalPlan::Join(join))
}
+ /// Coerce ASOF equality and ordered match expressions across input schemas.
+ pub fn coerce_asof_join(&mut self, mut join: AsOfJoin) -> Result<LogicalPlan> {
+ join.on = join
+ .on
+ .into_iter()
+ .map(|(left, right)| {
+ self.coerce_binary_op(
+ left,
+ join.left.schema(),
+ Operator::Eq,
+ right,
+ join.right.schema(),
+ )
+ })
+ .collect::<Result<_>>()?;
+ let (left, right) = self.coerce_binary_op(
+ join.match_condition.left,
+ join.left.schema(),
+ join.match_condition.op,
+ join.match_condition.right,
+ join.right.schema(),
+ )?;
+ join.match_condition = Box::new(AsOfMatch {
+ left,
+ op: join.match_condition.op,
+ right,
+ });
+ Ok(LogicalPlan::AsOfJoin(join))
+ }
+
/// Coerce the union’s inputs to a common schema compatible with all inputs.
/// This occurs after wildcard expansion and the coercion of the input expressions.
pub fn coerce_union(union_plan: Union) -> Result<LogicalPlan> {
diff --git a/datafusion/optimizer/src/common_subexpr_eliminate.rs b/datafusion/optimizer/src/common_subexpr_eliminate.rs
index 94d61b7487..6004ccb86e 100644
--- a/datafusion/optimizer/src/common_subexpr_eliminate.rs
+++ b/datafusion/optimizer/src/common_subexpr_eliminate.rs
@@ -566,6 +566,7 @@ impl OptimizerRule for CommonSubexprEliminate {
LogicalPlan::Window(window) => self.try_optimize_window(window, config)?,
LogicalPlan::Aggregate(agg) => self.try_optimize_aggregate(agg, config)?,
LogicalPlan::Join(_)
+ | LogicalPlan::AsOfJoin(_)
| LogicalPlan::Repartition(_)
| LogicalPlan::Union(_)
| LogicalPlan::TableScan(_)
diff --git a/datafusion/optimizer/src/optimize_projections/mod.rs b/datafusion/optimizer/src/optimize_projections/mod.rs
index 9604bf2f18..3bbaf887ca 100644
--- a/datafusion/optimizer/src/optimize_projections/mod.rs
+++ b/datafusion/optimizer/src/optimize_projections/mod.rs
@@ -407,6 +407,26 @@ fn optimize_projections(
right_indices.with_projection_beneficial(),
]
}
+ LogicalPlan::AsOfJoin(join) => {
+ let left_len = join.left.schema().fields().len();
+ let mut left_required = Vec::new();
+ let mut right_required = Vec::new();
+ for index in indices.indices() {
+ if *index < left_len {
+ left_required.push(*index);
+ } else {
+ right_required.push(*index - left_len);
+ }
+ }
+ let left_indices = RequiredIndices::new_from_indices(left_required)
+ .with_plan_exprs(&plan, join.left.schema())?;
+ let right_indices = RequiredIndices::new_from_indices(right_required)
+ .with_plan_exprs(&plan, join.right.schema())?;
+ vec![
+ left_indices.with_projection_beneficial(),
+ right_indices.with_projection_beneficial(),
+ ]
+ }
// these nodes are explicitly rewritten in the match statement above
LogicalPlan::Projection(_)
| LogicalPlan::Aggregate(_)
diff --git a/datafusion/optimizer/src/optimizer.rs b/datafusion/optimizer/src/optimizer.rs
index 46c0118095..ca4a6688b5 100644
--- a/datafusion/optimizer/src/optimizer.rs
+++ b/datafusion/optimizer/src/optimizer.rs
@@ -411,6 +411,11 @@ fn map_children_mut<F: FnMut(&mut LogicalPlan) -> Result<bool>>(
let r = f(Arc::make_mut(right))?;
l || r
}
+ LogicalPlan::AsOfJoin(join) => {
+ let l = f(Arc::make_mut(&mut join.left))?;
+ let r = f(Arc::make_mut(&mut join.right))?;
+ l || r
+ }
LogicalPlan::Union(Union { inputs, .. }) => {
let mut changed = false;
for input in inputs {
diff --git a/datafusion/physical-plan/src/joins/asof_join.rs b/datafusion/physical-plan/src/joins/asof_join.rs
index 7cc19e1153..3e67e929d0 100644
--- a/datafusion/physical-plan/src/joins/asof_join.rs
+++ b/datafusion/physical-plan/src/joins/asof_join.rs
@@ -169,6 +169,10 @@ impl AsOfJoinExec {
/// floating-point equality keys are not supported because Arrow sorting
/// distinguishes signed zero while SQL equality does not. Projection indices
/// refer to the full left-then-right join schema.
+ ///
+ /// The logical ASOF constructor validates the corresponding pre-coercion
+ /// contract. Keep the shared operator, side-ownership, and determinism checks
+ /// aligned across both public entry points.
pub fn try_new(
left: Arc<dyn ExecutionPlan>,
right: Arc<dyn ExecutionPlan>,
diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs
index 7fa6fd4ed5..90d48b056c 100644
--- a/datafusion/proto/src/logical_plan/mod.rs
+++ b/datafusion/proto/src/logical_plan/mod.rs
@@ -2225,6 +2225,9 @@ impl AsLogicalPlan for LogicalPlanNode {
LogicalPlan::DescribeTable(_) => Err(proto_error(
"LogicalPlan serde is not yet implemented for DescribeTable",
)),
+ LogicalPlan::AsOfJoin(_) => Err(proto_error(
+ "LogicalPlan serde is not yet implemented for AsOfJoin",
+ )),
LogicalPlan::RecursiveQuery(recursive) => {
let static_term = LogicalPlanNode::try_from_logical_plan(
recursive.static_term.as_ref(),
diff --git a/datafusion/sql/src/unparser/plan.rs b/datafusion/sql/src/unparser/plan.rs
index 1a4a094e96..29d4d1b5b8 100644
--- a/datafusion/sql/src/unparser/plan.rs
+++ b/datafusion/sql/src/unparser/plan.rs
@@ -204,6 +204,7 @@ impl Unparser<'_> {
| LogicalPlan::Copy(_)
| LogicalPlan::DescribeTable(_)
| LogicalPlan::RecursiveQuery(_)
+ | LogicalPlan::AsOfJoin(_)
| LogicalPlan::Unnest(_) => not_impl_err!("Unsupported plan: {plan:?}"),
}
}
diff --git a/datafusion/substrait/src/logical_plan/producer/rel/mod.rs b/datafusion/substrait/src/logical_plan/producer/rel/mod.rs
index c3599a2635..15f59919a2 100644
--- a/datafusion/substrait/src/logical_plan/producer/rel/mod.rs
+++ b/datafusion/substrait/src/logical_plan/producer/rel/mod.rs
@@ -51,6 +51,9 @@ pub fn to_substrait_rel(
LogicalPlan::Aggregate(plan) => producer.handle_aggregate(plan),
LogicalPlan::Sort(plan) => producer.handle_sort(plan),
LogicalPlan::Join(plan) => producer.handle_join(plan),
+ LogicalPlan::AsOfJoin(plan) => {
+ not_impl_err!("Substrait ASOF join is not supported: {plan:?}")?
+ }
LogicalPlan::Repartition(plan) => producer.handle_repartition(plan),
LogicalPlan::Union(plan) => producer.handle_union(plan),
LogicalPlan::TableScan(plan) => producer.handle_table_scan(plan),
diff --git a/datafusion/substrait/tests/cases/serialize.rs b/datafusion/substrait/tests/cases/serialize.rs
index 1981ef66db..75e1fe251a 100644
--- a/datafusion/substrait/tests/cases/serialize.rs
+++ b/datafusion/substrait/tests/cases/serialize.rs
@@ -18,7 +18,7 @@
#[cfg(test)]
mod tests {
use datafusion::datasource::provider_as_source;
- use datafusion::logical_expr::LogicalPlanBuilder;
+ use datafusion::logical_expr::{AsOfMatch, LogicalPlanBuilder, Operator};
use datafusion_substrait::logical_plan::consumer::from_substrait_plan;
use datafusion_substrait::logical_plan::producer::to_substrait_plan;
use datafusion_substrait::serializer;
@@ -27,7 +27,7 @@ mod tests {
use datafusion::prelude::*;
use insta::assert_snapshot;
- use std::fs;
+ use std::{fs, sync::Arc};
use substrait::proto::expression::field_reference::{ReferenceType, RootType};
use substrait::proto::expression::reference_segment;
use substrait::proto::expression::{ReferenceSegment, RexType};
@@ -103,6 +103,29 @@ mod tests {
Ok(())
}
+ #[tokio::test]
+ async fn asof_join_fails_closed_until_substrait_has_an_extension() -> Result<()> {
+ let ctx = create_context().await?;
+ let table = provider_as_source(ctx.table_provider("data").await?);
+ let left = LogicalPlanBuilder::scan("l", Arc::clone(&table), None)?.build()?;
+ let right = LogicalPlanBuilder::scan("r", table, None)?.build()?;
+ let plan = LogicalPlanBuilder::from(left)
+ .asof_join(
+ right,
+ vec![(col("l.b"), col("r.b"))],
+ AsOfMatch::new(col("l.a"), Operator::GtEq, col("r.a")),
+ )?
+ .build()?;
+ let error = to_substrait_plan(&plan, &ctx.state())
+ .expect_err("ASOF must not be lowered to a generic Substrait join");
+ assert!(
+ error
+ .to_string()
+ .contains("Substrait ASOF join is not supported")
+ );
+ Ok(())
+ }
+
#[tokio::test]
async fn include_remaps_for_projects() -> Result<()> {
let ctx = create_context().await?;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org