You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "wirybeaver (via GitHub)" <gi...@apache.org> on 2026/06/17 01:22:41 UTC
[PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
wirybeaver opened a new pull request, #22988:
URL: https://github.com/apache/datafusion/pull/22988
## Which issue does this PR close?
Follow-up to #20763 (merged) which added `MergeIntoOp`, `MergeIntoClause`, and proto types.
## Rationale for this change
`MERGE INTO` (SQL:2003) is a widely-used DML statement for upsert/conditional update workloads. This PR wires the types introduced in #20763 through the SQL planner, physical planner, and `TableProvider` trait so that table implementations can actually execute merge operations.
## What changes are included in this PR?
**`datafusion/catalog` — `TableProvider` trait extension**
- Add `merge_into(source, on, clauses)` async method with a default `not_impl_err` impl so existing providers are unaffected.
**`datafusion/sql` — SQL → LogicalPlan**
- `statement.rs`: parse `Statement::Merge` into `LogicalPlan::Dml` with `WriteOp::MergeInto`.
- Resolve the target table and plan the `USING` source into a `LogicalPlan`.
- Build a combined target+source schema to resolve `ON` and `WHEN` expressions.
- Convert `ON` condition and `WHEN MATCHED / NOT MATCHED` clauses to DataFusion `Expr`.
**`datafusion/expr` — expression plumbing**
- `MergeIntoOp::exprs()`: stable iteration over all expressions (ON, then per-clause predicate + action values).
- `MergeIntoOp::with_new_exprs()`: rebuild op from a transformed expr vector.
- Branch `LogicalPlan::apply_expressions`, `map_expressions`, and `with_new_exprs` on `WriteOp::MergeInto` so optimizers can rewrite merge expressions. Other `WriteOp` variants are unchanged.
**`datafusion/core` — physical planner dispatch**
- Dispatch `WriteOp::MergeInto` in the physical planner.
- Recover the `TableProvider` via `source_as_provider()`, extract the source `ExecutionPlan`, and call `TableProvider::merge_into`.
## Are these changes tested?
- The SQL planner path is exercised by `datafusion/proto/tests/cases/roundtrip_logical_plan.rs` (proto round-trip for `MergeInto`).
- Unit tests for `MergeIntoOp::exprs` / `with_new_exprs` are included in `dml.rs`.
- End-to-end integration tests require a `TableProvider` that implements `merge_into`; that is left to follow-up once a concrete provider (e.g. Delta Lake) adopts the hook.
## Are there any user-facing changes?
- **`TableProvider`** gains a new `merge_into` method. The default implementation returns `not_impl_err`, so existing implementations compile without changes.
- `MERGE INTO <table> USING <source> ON <cond> WHEN ...` SQL syntax is now accepted by the DataFusion SQL parser and planner.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-4964992110
hi @wirybeaver
Can you resolve the merge conflicts?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #22988:
URL: https://github.com/apache/datafusion/pull/22988
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3622336477
##########
datafusion/sql/src/statement.rs:
##########
@@ -2406,6 +2411,252 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
+ Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
Review Comment:
Could we handle a collision between the canonical target qualifier and the source alias here?
For example:
```sql
MERGE INTO target AS t
USING source AS target
ON t.id = target.id
WHEN MATCHED THEN DELETE
```
Planning initially resolves `t.id` and `target.id` as distinct columns. This canonicalization then rewrites the target reference to `target.id`, so both references are stored with the same qualifier. That irreversibly collapses the target and source namespaces.
Later, analysis merges the source schema before the reconstructed target schema. As a result, qualified lookup may resolve both references to the source column, turning the condition into a comparison of the source column with itself instead of comparing the target and source values.
Please preserve a collision-free qualifier for target resolution, or explicitly reject this alias collision until the logical representation can retain enough resolution context.
Could you also add a `SessionContext::sql()` regression test where the target and source have identically named columns and the source alias matches the target table's real name? The test should either verify an explicit rejection or confirm that the analyzed `MergeIntoOp` still distinguishes the target and source references. Checking only that planning succeeds would not catch silent misresolution.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5120290920
@alamb This PR passes all test and is ready to merge.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5222696095
> Thanks @wirybeaver -- I am sorry for the delay
>
> I think some SLT tests will help thsi too (I will make a follow on PR)
>
> Can you file some follow on tickets for the following follow on items?
>
> 1. EXPLAIN shows nothing about the merge (e.g. no ON conditions, WHEN clause, etc)
> 2. Update `docs/source/user-guide/sql/dml.md` to mention MERGE INTO.
>
> We may also want to consider implementing a basic MERGE INTO for MemTable (keep it as simple as possible to show the design)
Thanks for the guidance. Will follow your suggestion to create a follow up PR. Moreover, I also wanted to address the two edge cases that I mentioned in the summary caused by the target alias replacement.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5221599102
Thanks also @kosiew for helping to get this PR into good shape to merge
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3477731404
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
Review Comment:
Fixed — when an alias is present we now use it as the schema qualifier via `TableReference::bare(alias.name.value.clone())`, so `t.col` resolves correctly in `ON` and `WHEN` expressions.
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_table_ref.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ ));
+
+ Ok(plan)
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
+ AssignmentTarget::ColumnName(cols) => cols
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column id"))?
+ .as_ident()
+ .unwrap()
+ .value
+ .clone(),
+ _ => plan_err!("Tuples are not supported")?,
+ };
+ // Validate column exists in target
+ target_schema.field_with_unqualified_name(&col_name)?;
+ let value = self.sql_to_expr(
+ assign.value,
+ combined_schema,
+ planner_context,
+ )?;
+ Ok((col_name, value))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ MergeIntoAction::Update(assignments)
+ }
+ ast::MergeAction::Insert(insert_expr) => {
+ let columns: Vec<String> = insert_expr
Review Comment:
Fixed — we now validate before building the action: duplicate column names are rejected, each named column is checked against the target schema via `field_with_unqualified_name`, and the value count must equal either the explicit column count (when a column list is given) or the full target schema width (when none is given).
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_table_ref.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
Review Comment:
Done, returns `Ok(LogicalPlan::Dml(...))` directly now.
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_table_ref.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ ));
+
+ Ok(plan)
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
Review Comment:
Done — extracted `ident_from_object_name_last` as a private associated function that returns `Result<String>` and errors on non-ident parts. Both UPDATE and INSERT column extraction now go through it.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3650019540
##########
datafusion/sql/src/statement.rs:
##########
@@ -2414,6 +2419,403 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ if clauses.is_empty() {
+ return plan_err!("MERGE INTO requires at least one WHEN clause");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match table {
+ TableFactor::Table {
+ name,
+ alias,
+ args,
+ with_hints,
+ version,
+ with_ordinality,
+ partitions,
+ json_path,
+ sample,
+ index_hints,
+ } => {
+ if alias
+ .as_ref()
+ .is_some_and(|alias| !alias.columns.is_empty())
+ {
+ return not_impl_err!(
+ "MERGE target alias column lists are not supported"
+ );
+ }
+ if args.is_some()
+ || !with_hints.is_empty()
+ || version.is_some()
+ || with_ordinality
+ || !partitions.is_empty()
+ || json_path.is_some()
+ || sample.is_some()
+ || !index_hints.is_empty()
+ {
+ return not_impl_err!(
+ "MERGE target table modifiers are not supported"
+ );
+ }
+ (name, alias)
+ }
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_qualifier,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ // Target references in correlated subqueries are represented as
+ // `OuterReferenceColumn`s inside the embedded logical plan. The
+ // alias canonicalization below only rewrites top-level expression
+ // columns, so accepting such a subquery would leave the target
+ // alias in the public MERGE representation. Reject this case until
+ // the alias can be rewritten scope-safely inside subquery plans.
+ for expr in merge_op.exprs() {
+ if Self::has_outer_reference_to_qualifier(expr, &target_qualifier)? {
+ return not_impl_err!(
+ "MERGE subqueries correlated to target alias \
+ '{target_qualifier}' are not supported"
+ );
+ }
+ }
+
+ // Canonicalizing target columns to `target_table_ref` is only safe
+ // when the source does not already use that qualifier. If it does
+ // (e.g. `MERGE INTO target AS t USING source AS target`), the two
+ // namespaces would collapse and later resolution could silently
+ // pick the source column for a target reference. Reject that
+ // collision rather than change the meaning of the condition.
+ if source_plan.schema().iter().any(|(qualifier, _)| {
+ qualifier.is_some_and(|q| q.resolved_eq(&target_table_ref))
+ }) {
+ return plan_err!(
+ "MERGE source may not use the target table name '{target_table_ref}' \
+ as a qualifier while the target is aliased as '{target_qualifier}'; \
+ use a different source alias"
+ );
+ }
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
+ Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
+ Transformed::yes(Expr::Column(Column::new(Some(to.clone()), col.name))),
+ ),
+ other => Ok(Transformed::no(other)),
+ })
+ .map(|transformed| transformed.data)
+ }
+
+ /// Return true if an expression contains a subquery whose embedded plan
+ /// has an outer reference qualified by `qualifier`.
+ fn has_outer_reference_to_qualifier(
+ expr: &Expr,
+ qualifier: &TableReference,
+ ) -> Result<bool> {
+ let mut found = false;
+ expr.apply(|expr| {
+ let subquery = match expr {
+ Expr::Exists(exists) => Some(&exists.subquery),
+ Expr::InSubquery(in_subquery) => Some(&in_subquery.subquery),
+ Expr::SetComparison(set_comparison) => Some(&set_comparison.subquery),
+ Expr::ScalarSubquery(subquery) => Some(subquery),
+ _ => None,
+ };
+
+ if let Some(subquery) = subquery {
+ subquery.subquery.apply_with_subqueries(|plan| {
+ plan.apply_expressions(|expr| {
+ expr.apply(|expr| {
+ if let Expr::OuterReferenceColumn(_, column) = expr
+ && column.relation.as_ref() == Some(qualifier)
Review Comment:
`has_outer_reference_to_qualifier` currently decides whether a subquery is correlated to the MERGE target by comparing the `OuterReferenceColumn` qualifier with the target alias. The problem is that qualifiers are scope local, so the same alias can legally refer to a relation introduced inside a nested subquery.
For example, this valid query is currently rejected:
```sql
MERGE INTO target AS t
USING source AS s
ON EXISTS (
SELECT 1
FROM source AS t
WHERE EXISTS (
SELECT 1 FROM source AS x WHERE x.id = t.id
)
)
WHEN MATCHED THEN DELETE
```
The deepest `outer_ref(t.id)` refers to the immediately enclosing `source AS t`, not the MERGE target, but it is still treated as target correlation and produces:
```text
This feature is not implemented: MERGE subqueries correlated to target alias 't' are not supported
```
This is a pretty narrow false positive, and there is an easy workaround by using a different inner alias such as `source AS tt`. With that change, planning succeeds and reaches `TableProvider::merge_into`, where the expected Base provider `NotImplemented` error is returned.
I think the guard should become scope aware by accounting for aliases introduced by each embedded plan. It would also be good to add this nested alias shadowing case to the `SessionContext::sql()` regression tests so we do not accidentally regress it later.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3623797024
##########
datafusion/sql/src/statement.rs:
##########
@@ -2406,6 +2411,252 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
+ Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
Review Comment:
Thanks — you're right, and I confirmed it before fixing. I reproduced the exact scenario via `SessionContext::sql()`: for `MERGE INTO target AS t USING source AS target ON t.id = target.id`, the stored condition collapsed to `target.id = target.id` and the plan analyzed silently, so `t.id = target.id` did become a source-vs-source comparison. Good catch.
Fix: canonicalizing the aliased target reference to `target_table_ref` is inherently lossy when the source already uses that qualifier, so rather than silently change the condition's meaning I now reject that collision during planning:
```rust
if target_qualifier != target_table_ref {
// Canonicalizing target columns to `target_table_ref` is only safe when
// the source does not already use that qualifier ...
if source_plan
.schema()
.iter()
.any(|(qualifier, _)| qualifier == Some(&target_table_ref))
{
return plan_err!(
"MERGE source may not use the target table name '{target_table_ref}' \
as a qualifier while the target is aliased as '{target_qualifier}'; \
use a different source alias"
);
}
// ... canonicalize
}
```
Regression tests (both assert explicit rejection, per your note that "checking only that planning succeeds would not catch silent misresolution"):
- `merge_into_rejects_source_alias_colliding_with_target_name` — a `SessionContext::sql()` test in `datafusion/core/tests/sql/sql_api.rs`. Target and source both have an identically named `id` column and the source is aliased to the target table's real name (`USING source AS target`).
- `plan_merge_into_rejects_source_alias_colliding_with_target_name` — the same collision at the SQL-planner level for fast coverage.
The non-colliding aliased case still canonicalizes as before, covered by `plan_merge_into_canonicalizes_target_alias`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3643613171
##########
datafusion/sql/src/statement.rs:
##########
@@ -2406,6 +2411,269 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ // Canonicalizing target columns to `target_table_ref` is only safe
+ // when the source does not already use that qualifier. If it does
+ // (e.g. `MERGE INTO target AS t USING source AS target`), the two
+ // namespaces would collapse and later resolution could silently
+ // pick the source column for a target reference. Reject that
+ // collision rather than change the meaning of the condition.
+ if source_plan
+ .schema()
+ .iter()
+ .any(|(qualifier, _)| qualifier == Some(&target_table_ref))
+ {
+ return plan_err!(
+ "MERGE source may not use the target table name '{target_table_ref}' \
+ as a qualifier while the target is aliased as '{target_qualifier}'; \
+ use a different source alias"
+ );
+ }
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
+ Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
+ Transformed::yes(Expr::Column(Column::new(Some(to.clone()), col.name))),
+ ),
+ other => Ok(Transformed::no(other)),
+ })
+ .map(|transformed| transformed.data)
+ }
+
+ fn ident_from_object_name_last(name: &ObjectName) -> Result<Ident> {
+ let part = name
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column name"))?;
+ part.as_ident()
+ .cloned()
+ .ok_or_else(|| plan_datafusion_err!("Expected simple identifier"))
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
Review Comment:
Thanks for catching this. Addressed in 8de4f2851: the planner now rejects UPDATE WHERE, UPDATE ... DELETE WHERE, and INSERT WHERE rather than silently dropping them. I also added table-driven regressions for all three forms.
##########
datafusion/sql/src/statement.rs:
##########
@@ -2406,6 +2411,269 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ // Canonicalizing target columns to `target_table_ref` is only safe
+ // when the source does not already use that qualifier. If it does
+ // (e.g. `MERGE INTO target AS t USING source AS target`), the two
+ // namespaces would collapse and later resolution could silently
+ // pick the source column for a target reference. Reject that
+ // collision rather than change the meaning of the condition.
+ if source_plan
+ .schema()
+ .iter()
+ .any(|(qualifier, _)| qualifier == Some(&target_table_ref))
+ {
+ return plan_err!(
+ "MERGE source may not use the target table name '{target_table_ref}' \
+ as a qualifier while the target is aliased as '{target_qualifier}'; \
+ use a different source alias"
+ );
+ }
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
Review Comment:
Thanks, you're right. Addressed in 8de4f2851: MERGE subqueries correlated to an aliased target are now explicitly rejected until we can rewrite OuterReferenceColumn and subquery metadata scope-safely. The SessionContext regression covers this case and also verifies source-correlated and uncorrelated set-comparison subqueries still optimize successfully.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5067174764
Maintainer note: this update extends the public #[repr(C)] FFI_TableProvider layout with merge_into, so the PR needs the `api change` label. I attempted to add it, but my account does not have label permission.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5219876441
Checkung it out -- hope to get it reviewed and merged shortly
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3466496112
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_table_ref.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
Review Comment:
Small cleanup suggestion: this could return `Ok(LogicalPlan::Dml(DmlStatement::new(...)))` directly instead of binding `let plan` and then returning `Ok(plan)`.
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_table_ref.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ ));
+
+ Ok(plan)
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
+ AssignmentTarget::ColumnName(cols) => cols
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column id"))?
+ .as_ident()
+ .unwrap()
+ .value
+ .clone(),
+ _ => plan_err!("Tuples are not supported")?,
+ };
+ // Validate column exists in target
+ target_schema.field_with_unqualified_name(&col_name)?;
+ let value = self.sql_to_expr(
+ assign.value,
+ combined_schema,
+ planner_context,
+ )?;
+ Ok((col_name, value))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ MergeIntoAction::Update(assignments)
+ }
+ ast::MergeAction::Insert(insert_expr) => {
+ let columns: Vec<String> = insert_expr
Review Comment:
I think the `MERGE INSERT` actions need the same kind of validation that regular INSERT planning does. At the moment, unknown columns, duplicate columns, and mismatched `columns` / `values` lengths can all plan successfully. Empty `columns` also does not appear to require values for every target column.
That can leave TableProviders receiving malformed actions instead of a validated logical operation. Could we normalize and validate target column names, reject duplicates, and verify that the value count matches either the explicit columns or the full target schema?
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_table_ref.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ let plan = LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ ));
+
+ Ok(plan)
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
Review Comment:
Small follow-up suggestion: the update-column extraction and insert-column extraction both walk an `ObjectName` and take the last identifier. It might be worth adding a small private helper that returns `Result<String>` and produces a planner error for non-ident parts, rather than using `unwrap()` at each call site.
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,189 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
Review Comment:
I think target aliases need a bit more handling here. Right now `MERGE INTO target AS t USING source AS s ON t.id = s.id ...` accepts the alias, but `target_schema` is still qualified with the table name rather than `t`. That means normal alias-qualified references in the `ON` clause or `WHEN AND` expressions will not resolve.
Could we either reject target aliases explicitly for now, or qualify the target side with the alias so this matches the rest of SQL planning?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3607281710
##########
datafusion/expr/src/logical_plan/dml.rs:
##########
@@ -299,12 +299,129 @@ impl Display for InsertOp {
}
/// Describes a MERGE INTO operation's parameters.
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct MergeIntoOp {
/// The join condition from `ON <expr>`.
pub on: Expr,
/// The WHEN clauses, in the order they appeared in the SQL.
pub clauses: Vec<MergeIntoClause>,
+ /// Schema of the target table (qualified with the alias or table name used
+ /// in the SQL). Stored here so the analyzer can build a combined
+ /// target+source schema for type coercion and function rewrites.
+ pub target_schema: DFSchemaRef,
Review Comment:
Could we avoid adding `target_schema` as a new public field on `MergeIntoOp`?
Because `MergeIntoOp` is a public struct that downstream users can construct with struct literals, adding a required field is a SemVer-breaking change.
I do not think that break is necessary for this PR. `target_schema` looks like analyzer and planning context rather than essential public MERGE operation data. Exposing it here makes downstream callers provide internal state they should not need to know about.
Can we keep the additional planning state outside the publicly constructible struct, or introduce a SemVer-compatible constructor or internal representation before requiring new state from callers?
One possible approach is to keep `MergeIntoOp { on, clauses }`, derive the target schema in the analyzer, type coercion, or function rewrite passes from the existing `DmlStatement { table_name, target, .. }`, and canonicalize alias-qualified target references during SQL planning.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3534774298
##########
datafusion/expr/src/logical_plan/tree_node.rs:
##########
@@ -480,6 +480,10 @@ impl LogicalPlan {
}
_ => Ok(TreeNodeRecursion::Continue),
},
+ LogicalPlan::Dml(DmlStatement {
Review Comment:
I think we should avoid exposing `MergeIntoOp` expressions through generic logical-plan expression traversal for now.
The analyzer and type coercion rewrite schema is built from `plan.inputs()`, but the `Dml` node only has the `USING` source plan as input. MERGE `ON` and `WHEN` expressions are planned against both the target and source schemas, so a query like `MERGE INTO t USING s ON t.id = s.id` can fail during `SessionContext::sql()` analysis because `t.id` is not present in the source-only schema.
Could we either keep these expressions out of generic traversal for the initial implementation, or store enough target schema and qualifier information on the MERGE operation so analyzer/type coercion can rewrite them against the combined target plus source schema?
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,213 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier,
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ )))
+ }
+
+ fn ident_from_object_name_last(name: &ObjectName) -> Result<String> {
+ let part = name
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column name"))?;
+ part.as_ident()
+ .ok_or_else(|| plan_datafusion_err!("Expected simple identifier"))
+ .map(|ident| ident.value.clone())
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
+ AssignmentTarget::ColumnName(cols) => {
+ let raw = Self::ident_from_object_name_last(cols)?;
+ self.ident_normalizer.normalize(Ident::new(raw))
+ }
+ _ => plan_err!("Tuples are not supported")?,
+ };
+ // Validate column exists in target
+ target_schema.field_with_unqualified_name(&col_name)?;
+ let value = self.sql_to_expr(
+ assign.value,
+ combined_schema,
+ planner_context,
+ )?;
+ Ok((col_name, value))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ MergeIntoAction::Update(assignments)
+ }
+ ast::MergeAction::Insert(insert_expr) => {
+ let columns: Vec<String> = insert_expr
+ .columns
+ .iter()
+ .map(|c| {
+ let raw = Self::ident_from_object_name_last(c)?;
+ Ok(self.ident_normalizer.normalize(Ident::new(raw)))
Review Comment:
Thanks for working on this one. I think the current fix still has a quoted-identifier edge case.
The UPDATE/INSERT column extraction now returns only `Ident.value`, then reconstructs it with `Ident::new(raw)` before normalization. That loses `quote_style`, so quoted identifiers get normalized as if they were unquoted. For example, `INSERT ("ID")` or `UPDATE SET "ID" = ...` can become `id`, fail to match a target column named `ID`, or be treated as a duplicate of unquoted `ID`.
Regular INSERT preserves the original `Ident` and passes that to `ident_normalizer.normalize`. Could we make this helper return the last `Ident`, or otherwise preserve the quote style, and normalize the original ident before duplicate checks and schema lookup?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-4970279971
<!-- semver-check-comment -->
Thank you for opening this pull request!
Reviewer note: [cargo-semver-checks](https://github.com/obi1kenobi/cargo-semver-checks) reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).
<details>
<summary>Details</summary>
```
Cloning apache/main
Building datafusion v54.0.0 (current)
Built [ 118.159s] (current)
Parsing datafusion v54.0.0 (current)
Parsed [ 0.036s] (current)
Building datafusion v54.0.0 (baseline)
Built [ 107.862s] (baseline)
Parsing datafusion v54.0.0 (baseline)
Parsed [ 0.036s] (baseline)
Checking datafusion v54.0.0 -> v54.0.0 (no change; assume patch)
Checked [ 0.605s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 228.654s] datafusion
Building datafusion-catalog v54.0.0 (current)
Built [ 40.399s] (current)
Parsing datafusion-catalog v54.0.0 (current)
Parsed [ 0.025s] (current)
Building datafusion-catalog v54.0.0 (baseline)
Built [ 41.198s] (baseline)
Parsing datafusion-catalog v54.0.0 (baseline)
Parsed [ 0.028s] (baseline)
Checking datafusion-catalog v54.0.0 -> v54.0.0 (no change; assume patch)
Checked [ 0.161s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 82.815s] datafusion-catalog
Building datafusion-expr v54.0.0 (current)
Built [ 28.576s] (current)
Parsing datafusion-expr v54.0.0 (current)
Parsed [ 0.077s] (current)
Building datafusion-expr v54.0.0 (baseline)
Built [ 28.510s] (baseline)
Parsing datafusion-expr v54.0.0 (baseline)
Parsed [ 0.079s] (baseline)
Checking datafusion-expr v54.0.0 -> v54.0.0 (no change; assume patch)
Checked [ 1.363s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip
--- failure constructible_struct_adds_field: externally-constructible struct adds field ---
Description:
A pub struct constructible with a struct literal has a new pub field. Existing struct literals must be updated to include the new field.
ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.48.0/src/lints/constructible_struct_adds_field.ron
Failed in:
field MergeIntoOp.target_schema in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/dml.rs:311
field MergeIntoOp.target_schema in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/dml.rs:311
field MergeIntoOp.target_schema in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/dml.rs:311
field MergeIntoOp.target_schema in /home/runner/work/datafusion/datafusion/datafusion/expr/src/logical_plan/dml.rs:311
Summary semver requires new major version: 1 major and 0 minor checks failed
Finished [ 59.576s] datafusion-expr
Building datafusion-optimizer v54.0.0 (current)
Built [ 28.622s] (current)
Parsing datafusion-optimizer v54.0.0 (current)
Parsed [ 0.030s] (current)
Building datafusion-optimizer v54.0.0 (baseline)
Built [ 27.854s] (baseline)
Parsing datafusion-optimizer v54.0.0 (baseline)
Parsed [ 0.032s] (baseline)
Checking datafusion-optimizer v54.0.0 -> v54.0.0 (no change; assume patch)
Checked [ 0.182s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 57.507s] datafusion-optimizer
Building datafusion-proto v54.0.0 (current)
Built [ 62.873s] (current)
Parsing datafusion-proto v54.0.0 (current)
Parsed [ 0.018s] (current)
Building datafusion-proto v54.0.0 (baseline)
Built [ 61.908s] (baseline)
Parsing datafusion-proto v54.0.0 (baseline)
Parsed [ 0.018s] (baseline)
Checking datafusion-proto v54.0.0 -> v54.0.0 (no change; assume patch)
Checked [ 0.248s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 126.452s] datafusion-proto
Building datafusion-sql v54.0.0 (current)
Built [ 43.432s] (current)
Parsing datafusion-sql v54.0.0 (current)
Parsed [ 0.031s] (current)
Building datafusion-sql v54.0.0 (baseline)
Built [ 43.840s] (baseline)
Parsing datafusion-sql v54.0.0 (baseline)
Parsed [ 0.033s] (baseline)
Checking datafusion-sql v54.0.0 -> v54.0.0 (no change; assume patch)
Checked [ 0.256s] 223 checks: 223 pass, 30 skip
Summary no semver update required
Finished [ 88.955s] datafusion-sql
```
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5023568789
## [Codecov](https://app.codecov.io/gh/apache/datafusion/pull/22988?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `83.50254%` with `65 lines` in your changes missing coverage. Please review.
:warning: Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`main@3a29d6b`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#section-missing-base-commit) about missing BASE report.
| [Files with missing lines](https://app.codecov.io/gh/apache/datafusion/pull/22988?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [datafusion/sql/src/statement.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Fsql%2Fsrc%2Fstatement.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9zcWwvc3JjL3N0YXRlbWVudC5ycw==) | 80.32% | [16 Missing and 20 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...afusion/optimizer/src/analyzer/function\_rewrite.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Foptimizer%2Fsrc%2Fanalyzer%2Ffunction_rewrite.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9vcHRpbWl6ZXIvc3JjL2FuYWx5emVyL2Z1bmN0aW9uX3Jld3JpdGUucnM=) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/expr/src/logical\_plan/plan.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Fexpr%2Fsrc%2Flogical_plan%2Fplan.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9leHByL3NyYy9sb2dpY2FsX3BsYW4vcGxhbi5ycw==) | 0.00% | [7 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/optimizer/src/analyzer/type\_coercion.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Foptimizer%2Fsrc%2Fanalyzer%2Ftype_coercion.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9vcHRpbWl6ZXIvc3JjL2FuYWx5emVyL3R5cGVfY29lcmNpb24ucnM=) | 88.37% | [1 Missing and 4 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/core/src/physical\_planner.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Fcore%2Fsrc%2Fphysical_planner.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9jb3JlL3NyYy9waHlzaWNhbF9wbGFubmVyLnJz) | 72.72% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/expr/src/logical\_plan/tree\_node.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Fexpr%2Fsrc%2Flogical_plan%2Ftree_node.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9leHByL3NyYy9sb2dpY2FsX3BsYW4vdHJlZV9ub2RlLnJz) | 83.33% | [0 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [datafusion/catalog/src/table.rs](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&filepath=datafusion%2Fcatalog%2Fsrc%2Ftable.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-ZGF0YWZ1c2lvbi9jYXRhbG9nL3NyYy90YWJsZS5ycw==) | 0.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/datafusion/pull/22988?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #22988 +/- ##
=======================================
Coverage ? 80.70%
=======================================
Files ? 1089
Lines ? 368434
Branches ? 368434
=======================================
Hits ? 297347
Misses ? 53360
Partials ? 17727
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/datafusion/pull/22988?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5071141100
I split the FFI-specific changes into draft #23867. This PR now contains only the MERGE INTO planner, analyzer/optimizer, physical-planner, and `TableProvider` API work; it no longer changes the `FFI_TableProvider` ABI.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3489489677
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,207 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| TableReference::bare(a.name.value.clone()))
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier,
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ )))
+ }
+
+ fn ident_from_object_name_last(name: &ObjectName) -> Result<String> {
+ let part = name
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column name"))?;
+ part.as_ident()
+ .ok_or_else(|| plan_datafusion_err!("Expected simple identifier"))
+ .map(|ident| ident.value.clone())
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
+ AssignmentTarget::ColumnName(cols) => {
+ Self::ident_from_object_name_last(cols)?
+ }
+ _ => plan_err!("Tuples are not supported")?,
+ };
+ // Validate column exists in target
+ target_schema.field_with_unqualified_name(&col_name)?;
+ let value = self.sql_to_expr(
+ assign.value,
+ combined_schema,
+ planner_context,
+ )?;
+ Ok((col_name, value))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ MergeIntoAction::Update(assignments)
+ }
+ ast::MergeAction::Insert(insert_expr) => {
+ let columns: Vec<String> = insert_expr
+ .columns
+ .iter()
+ .map(|c| Self::ident_from_object_name_last(c))
Review Comment:
I think the MERGE INSERT column handling still needs one more normalization step. The extracted column names currently keep the raw `Ident.value` from `ident_from_object_name_last`, so validation happens before SQL identifier normalization.
That means something like `MERGE ... INSERT (ID) VALUES (...)` can be rejected against a target column named `id`, and duplicate detection may not match regular INSERT behavior.
Could we normalize the extracted column identifiers before duplicate checks and before the `field_with_unqualified_name` lookup, matching regular INSERT planning?
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,207 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| TableReference::bare(a.name.value.clone()))
Review Comment:
Nice fix to use the target alias as the schema qualifier here. One remaining issue is that this uses `a.name.value.clone()` directly, so unquoted aliases are not normalized the same way as the rest of planning.
For example, `MERGE INTO target AS T ... ON T.id = s.id` can store the target qualifier as `T`, while expression planning normalizes `T.id` to `t.id`. That can still make alias-qualified target references fail.
Could we build the alias qualifier with `self.ident_normalizer.normalize(a.name.clone())` before passing it to `TableReference::bare(...)`?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3559944792
##########
datafusion/sql/src/statement.rs:
##########
@@ -2407,6 +2411,213 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier,
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the DmlStatement
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ })),
+ Arc::new(source_plan),
+ )))
+ }
+
+ fn ident_from_object_name_last(name: &ObjectName) -> Result<String> {
+ let part = name
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column name"))?;
+ part.as_ident()
+ .ok_or_else(|| plan_datafusion_err!("Expected simple identifier"))
+ .map(|ident| ident.value.clone())
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
+ let assignments = update_expr
+ .assignments
+ .into_iter()
+ .map(|assign| {
+ let col_name = match &assign.target {
+ AssignmentTarget::ColumnName(cols) => {
+ let raw = Self::ident_from_object_name_last(cols)?;
+ self.ident_normalizer.normalize(Ident::new(raw))
+ }
+ _ => plan_err!("Tuples are not supported")?,
+ };
+ // Validate column exists in target
+ target_schema.field_with_unqualified_name(&col_name)?;
+ let value = self.sql_to_expr(
+ assign.value,
+ combined_schema,
+ planner_context,
+ )?;
+ Ok((col_name, value))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ MergeIntoAction::Update(assignments)
+ }
+ ast::MergeAction::Insert(insert_expr) => {
+ let columns: Vec<String> = insert_expr
+ .columns
+ .iter()
+ .map(|c| {
+ let raw = Self::ident_from_object_name_last(c)?;
+ Ok(self.ident_normalizer.normalize(Ident::new(raw)))
Review Comment:
Good catch — fixed. `ident_from_object_name_last` now returns the last `Ident` (preserving `quote_style`) instead of a bare `String`. Both UPDATE and INSERT column extraction pass that `Ident` straight into `self.ident_normalizer.normalize()`, so quoted `"ID"` stays `ID` and unquoted `ID` normalizes to `id`, matching regular INSERT/UPDATE behavior.
##########
datafusion/expr/src/logical_plan/tree_node.rs:
##########
@@ -480,6 +480,10 @@ impl LogicalPlan {
}
_ => Ok(TreeNodeRecursion::Continue),
},
+ LogicalPlan::Dml(DmlStatement {
Review Comment:
Went with storing target schema info on the MERGE operation. Added `target_schema: DFSchemaRef` to `MergeIntoOp`, populated from the qualified target schema computed during SQL planning. `TypeCoercion` and `ApplyFunctionRewrites` now merge `merge_op.target_schema` into the schema they build from `plan.inputs()` (same pattern already used for `TableScan`'s source schema), so both target and source columns are resolvable when coercing MERGE `ON`/`WHEN` expressions. For proto, `target_schema` isn't serialized — it's reconstructed from `table_name` + `target.schema()` at deserialization time, the same way `Join`/`Union` reconstruct their schemas from inputs rather than serializing them.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3607281710
##########
datafusion/expr/src/logical_plan/dml.rs:
##########
@@ -299,12 +299,129 @@ impl Display for InsertOp {
}
/// Describes a MERGE INTO operation's parameters.
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct MergeIntoOp {
/// The join condition from `ON <expr>`.
pub on: Expr,
/// The WHEN clauses, in the order they appeared in the SQL.
pub clauses: Vec<MergeIntoClause>,
+ /// Schema of the target table (qualified with the alias or table name used
+ /// in the SQL). Stored here so the analyzer can build a combined
+ /// target+source schema for type coercion and function rewrites.
+ pub target_schema: DFSchemaRef,
Review Comment:
Could we avoid adding `target_schema` as a new public field on `MergeIntoOp`?
Because `MergeIntoOp` is a public struct that downstream users can construct with struct literals, adding a required field is a SemVer-breaking change. This also matches the issue reported by the cargo-semver-checks bot.
I do not think that break is necessary for this PR. `target_schema` looks like analyzer and planning context rather than essential public MERGE operation data. Exposing it here makes downstream callers provide internal state they should not need to know about.
Please keep the additional planning state outside the publicly constructible struct, or introduce a SemVer-compatible constructor or internal representation before requiring new state from callers. One possible approach is to keep `MergeIntoOp { on, clauses }`, derive the target schema in the analyzer, type coercion, or function rewrite passes from the existing `DmlStatement { table_name, target, .. }`, and canonicalize alias-qualified target references during SQL planning.
##########
datafusion/proto/src/logical_plan/mod.rs:
##########
@@ -1258,11 +1258,21 @@ impl AsLogicalPlan for LogicalPlanNode {
.build()
}
LogicalPlanType::Dml(dml_node) => {
- let write_op =
+ let table_name =
+ from_table_reference(dml_node.table_name.as_ref(), "DML ")?;
+ let target = to_table_source(&dml_node.target, ctx, extension_codec)?;
+ let mut write_op =
from_proto::parse_write_op(dml_node, ctx, extension_codec)?;
+ if let WriteOp::MergeInto(ref mut merge_op) = write_op {
+ merge_op.target_schema =
Review Comment:
I think this reconstruction loses the target alias used during SQL expression resolution.
For example, with `MERGE INTO target AS t USING source AS s ON t.id = s.id`, the expressions may still refer to `t.id` after a proto roundtrip. However, this code rebuilds `target_schema` using `table_name`, so the qualifier becomes `target` instead of `t`. A later analyzer, type coercion, or function rewrite pass that relies on this schema may then fail to resolve the target-side columns.
Could we either preserve the target resolution qualifier during serialization or reconstruct it from serialized plan state that contains the SQL alias? My preference would be to avoid serializing this additional state by canonicalizing target alias references to the real target table qualifier before storing the logical plan. That would allow proto deserialization to safely rebuild the schema from `DmlStatement.table_name` and `target.schema()`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3637075437
##########
datafusion/sql/src/statement.rs:
##########
@@ -2406,6 +2411,269 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ // Canonicalizing target columns to `target_table_ref` is only safe
+ // when the source does not already use that qualifier. If it does
+ // (e.g. `MERGE INTO target AS t USING source AS target`), the two
+ // namespaces would collapse and later resolution could silently
+ // pick the source column for a target reference. Reject that
+ // collision rather than change the meaning of the condition.
+ if source_plan
+ .schema()
+ .iter()
+ .any(|(qualifier, _)| qualifier == Some(&target_table_ref))
+ {
+ return plan_err!(
+ "MERGE source may not use the target table name '{target_table_ref}' \
+ as a qualifier while the target is aliased as '{target_qualifier}'; \
+ use a different source alias"
+ );
+ }
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
+ Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
+ Transformed::yes(Expr::Column(Column::new(Some(to.clone()), col.name))),
+ ),
+ other => Ok(Transformed::no(other)),
+ })
+ .map(|transformed| transformed.data)
+ }
+
+ fn ident_from_object_name_last(name: &ObjectName) -> Result<Ident> {
+ let part = name
+ .0
+ .iter()
+ .last()
+ .ok_or_else(|| plan_datafusion_err!("Empty column name"))?;
+ part.as_ident()
+ .cloned()
+ .ok_or_else(|| plan_datafusion_err!("Expected simple identifier"))
+ }
+
+ fn merge_clause_to_plan(
+ &self,
+ clause: ast::MergeClause,
+ combined_schema: &DFSchema,
+ target_schema: &DFSchema,
+ _target_alias: &Option<ast::TableAlias>,
+ planner_context: &mut PlannerContext,
+ ) -> Result<MergeIntoClause> {
+ let kind = match clause.clause_kind {
+ ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
+ ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
+ ast::MergeClauseKind::NotMatchedByTarget => {
+ MergeIntoClauseKind::NotMatchedByTarget
+ }
+ ast::MergeClauseKind::NotMatchedBySource => {
+ MergeIntoClauseKind::NotMatchedBySource
+ }
+ };
+
+ let predicate = clause
+ .predicate
+ .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
+ .transpose()?;
+
+ let action = match clause.action {
+ ast::MergeAction::Update(update_expr) => {
Review Comment:
It looks like `sqlparser` preserves action-level predicates on `MergeUpdateExpr` (`update_predicate` and `delete_predicate`) and `MergeInsertExpr` (`insert_predicate`), but this conversion only reads the assignments, columns, and values.
As a result, statements like `WHEN MATCHED THEN UPDATE SET ... WHERE false` are planned as unconditional updates, and INSERT `WHERE` predicates are dropped as well. That means accepted SQL can modify a different set of rows than the user requested.
Could we either preserve these predicates in `MergeIntoClause` / `MergeIntoAction` and pass them through to providers, or reject any non-`None` action-level predicate during planning until the logical representation supports them?
It would also be good to add regression tests covering UPDATE `WHERE`, UPDATE `... DELETE WHERE`, and INSERT `WHERE`.
##########
datafusion/sql/src/statement.rs:
##########
@@ -2406,6 +2411,269 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match &table {
+ TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_alias,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ // Canonicalizing target columns to `target_table_ref` is only safe
+ // when the source does not already use that qualifier. If it does
+ // (e.g. `MERGE INTO target AS t USING source AS target`), the two
+ // namespaces would collapse and later resolution could silently
+ // pick the source column for a target reference. Reject that
+ // collision rather than change the meaning of the condition.
+ if source_plan
+ .schema()
+ .iter()
+ .any(|(qualifier, _)| qualifier == Some(&target_table_ref))
+ {
+ return plan_err!(
+ "MERGE source may not use the target table name '{target_table_ref}' \
+ as a qualifier while the target is aliased as '{target_qualifier}'; \
+ use a different source alias"
+ );
+ }
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
Review Comment:
This rewrite only handles `Expr::Column`. Target references inside MERGE subqueries become `Expr::OuterReferenceColumn` in the embedded logical plan, so they keep the SQL alias instead of being canonicalized.
For example:
```sql
MERGE INTO target AS t
USING source AS s
ON EXISTS (SELECT 1 FROM source x WHERE x.id = t.id)
WHEN MATCHED THEN DELETE
```
After `SessionContext::sql()`, the nested subquery still contains `outer_ref(t.id)`, while ordinary target references have already been rewritten to `target.id`. Since `MergeIntoOp` does not retain the target alias or an alias-to-target mapping, providers cannot reliably resolve that qualifier from the public logical representation.
Could we either canonicalize target-qualified `OuterReferenceColumn`s inside embedded subquery plans using a scope-aware traversal, or reject target-correlated MERGE subqueries until the logical representation retains enough information to resolve them?
A regression test that inspects the analyzed `MergeIntoOp` and the nested subquery after `SessionContext::sql()` would help catch this. Checking only that planning succeeds would not detect this issue.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "raghav-reglobe (via GitHub)" <gi...@apache.org>.
raghav-reglobe commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-4878506556
Took this branch for a spin against a provider-side consumer (we're building an Iceberg SCD2 upsert on this hook), and I think there's one blocking issue the current tests don't reach: **any target-column reference in the `ON` or `WHEN` expressions fails analysis**, so a realistic MERGE can't execute through `SessionContext::sql()`.
Minimal repro (custom provider registered as `t`, MemTable as `batch`):
```sql
MERGE INTO t USING batch s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET val = s.val
```
```
Error: Context("type_coercion", SchemaError(FieldNotFound { field: Column { relation: Some(Bare { table: "t" }), name: "id" }, valid_fields: [s.id, s.val, …] }))
```
What's happening: the `Dml` node's only input is the planned USING source, but `MergeIntoOp`'s expressions reference the combined target×source namespace. Since this PR wires those expressions into the generic tree-node traversal (`tree_node.rs`), the analyzer's `TypeCoercion` pass rewrites them against the node's input schema — which is source-only — and errors on the first target reference. As a probe, `ON s.id = s.id` (source-only) sails through analysis and reaches `merge_into()` fine, which pins the root cause. Everything else held up nicely in testing, for what it's worth — a window-function subquery as the USING source (`ROW_NUMBER`/`LEAD`), multi-condition `ON`, clause predicates, and the INSERT validation all plan exactly as expected.
Two directions I can see:
1. **Minimal**: don't expose `MergeIntoOp` exprs through the generic traversal (keep `exprs()`/`with_new_exprs` as inherent methods only). The analyzer/optimizer then leave the node alone, and providers receive un-coerced expressions — acceptable for providers that do their own resolution, and it ships a working v1.
2. **Fuller**: record the target's resolution qualifier on `MergeIntoOp` (the alias when present, else the table reference — today the alias is dropped after planning), and teach `TypeCoercion` to resolve MergeInto exprs against target-schema ⋈ input-schema. Coercion on these expressions genuinely matters (e.g. an `Int32` target key against an `Int64` window output in `ON`), and the recorded qualifier also gives providers an explicit way to tell target references from source references instead of inferring by absence.
I'd lean toward 2, but 1 seems fine to unblock if you'd rather keep this PR small. Happy to share the test module I used — a `CaptureMergeProvider` in the style of `dml_planning.rs` with basic + SCD2-shaped (window-subquery USING, triple-condition ON, clause predicate) cases — if it's useful as a starting point for end-to-end coverage here. Also happy to put up a patch for either direction if that helps.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3614960602
##########
datafusion/expr/src/logical_plan/dml.rs:
##########
@@ -299,12 +299,129 @@ impl Display for InsertOp {
}
/// Describes a MERGE INTO operation's parameters.
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct MergeIntoOp {
/// The join condition from `ON <expr>`.
pub on: Expr,
/// The WHEN clauses, in the order they appeared in the SQL.
pub clauses: Vec<MergeIntoClause>,
+ /// Schema of the target table (qualified with the alias or table name used
+ /// in the SQL). Stored here so the analyzer can build a combined
+ /// target+source schema for type coercion and function rewrites.
+ pub target_schema: DFSchemaRef,
Review Comment:
Good call — done. I removed the `target_schema` field from `MergeIntoOp`, so it's back to `{ on, clauses }` and the struct is no longer SemVer-breaking to construct with a struct literal.
The target schema is now **derived on demand** in the analyzer passes (type coercion and function rewrites) from the existing `DmlStatement { table_name, target }`, mirroring how `TableScan` already merges its source schema in the same functions:
```rust
if let LogicalPlan::Dml(DmlStatement {
op: WriteOp::MergeInto(_),
table_name,
target,
..
}) = &plan
{
let target_schema =
DFSchema::try_from_qualified_schema(table_name.clone(), &target.schema())?;
schema.merge(&target_schema);
}
```
To make that derivation sufficient, SQL planning now canonicalizes target alias-qualified column references (`t.col`) to the real target table qualifier before storing the plan, so no alias/planning state leaks onto the public struct. Added a planner test (`plan_merge_into_canonicalizes_target_alias`) verifying `MERGE INTO j1 AS t USING j2 AS s ON t.j1_id = s.j2_id ...` produces `j1.j1_id = s.j2_id` (target alias rewritten, source alias preserved).
##########
datafusion/proto/src/logical_plan/mod.rs:
##########
@@ -1258,11 +1258,21 @@ impl AsLogicalPlan for LogicalPlanNode {
.build()
}
LogicalPlanType::Dml(dml_node) => {
- let write_op =
+ let table_name =
+ from_table_reference(dml_node.table_name.as_ref(), "DML ")?;
+ let target = to_table_source(&dml_node.target, ctx, extension_codec)?;
+ let mut write_op =
from_proto::parse_write_op(dml_node, ctx, extension_codec)?;
+ if let WriteOp::MergeInto(ref mut merge_op) = write_op {
+ merge_op.target_schema =
Review Comment:
Agreed, and I went with your preferred approach: canonicalize target alias references to the real target table qualifier during SQL planning, then avoid serializing any extra state.
`merge_to_plan` now rewrites target-side columns so `t.id` is stored as `target.id` (source aliases like `s.id` are left untouched). Because the stored expressions always use the `table_name` qualifier, there's nothing alias-specific left to lose across a proto roundtrip — so I removed this reconstruction block entirely. Proto deserialization just rebuilds the `DmlStatement` from `table_name` + `target`, and the analyzer derives the target schema from those on the fly when it needs to resolve MERGE expressions.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "wirybeaver (via GitHub)" <gi...@apache.org>.
wirybeaver commented on code in PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#discussion_r3650983740
##########
datafusion/sql/src/statement.rs:
##########
@@ -2414,6 +2419,403 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
Ok(plan)
}
+ fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
+ let ast::Merge {
+ table,
+ source,
+ on,
+ clauses,
+ into: _,
+ merge_token: _,
+ optimizer_hints,
+ output,
+ } = merge;
+
+ if !optimizer_hints.is_empty() {
+ plan_err!("Optimizer hints not supported")?;
+ }
+
+ if output.is_some() {
+ return not_impl_err!("MERGE OUTPUT clause is not supported");
+ }
+
+ if clauses.is_empty() {
+ return plan_err!("MERGE INTO requires at least one WHEN clause");
+ }
+
+ // 1. Resolve target table
+ let (target_table_name, target_alias) = match table {
+ TableFactor::Table {
+ name,
+ alias,
+ args,
+ with_hints,
+ version,
+ with_ordinality,
+ partitions,
+ json_path,
+ sample,
+ index_hints,
+ } => {
+ if alias
+ .as_ref()
+ .is_some_and(|alias| !alias.columns.is_empty())
+ {
+ return not_impl_err!(
+ "MERGE target alias column lists are not supported"
+ );
+ }
+ if args.is_some()
+ || !with_hints.is_empty()
+ || version.is_some()
+ || with_ordinality
+ || !partitions.is_empty()
+ || json_path.is_some()
+ || sample.is_some()
+ || !index_hints.is_empty()
+ {
+ return not_impl_err!(
+ "MERGE target table modifiers are not supported"
+ );
+ }
+ (name, alias)
+ }
+ _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
+ };
+ let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
+ let target_table_source = self
+ .context_provider
+ .get_table_source(target_table_ref.clone())?;
+ // Use alias as schema qualifier so `t.col` resolves when user writes
+ // `MERGE INTO target AS t`. Fall back to the table reference itself.
+ let target_qualifier = target_alias
+ .as_ref()
+ .map(|a| {
+ TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
+ })
+ .unwrap_or_else(|| target_table_ref.clone());
+ let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
+ target_qualifier.clone(),
+ &target_table_source.schema(),
+ )?);
+
+ // 2. Plan the source (USING clause) as a LogicalPlan
+ let mut planner_context = PlannerContext::new();
+ let source_table_with_joins = TableWithJoins {
+ relation: source,
+ joins: vec![],
+ };
+ let source_plan =
+ self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
+
+ // 3. Build a combined schema for resolving expressions in ON and WHEN clauses
+ let combined_schema =
+ Arc::new(target_schema.as_ref().join(source_plan.schema())?);
+
+ // 4. Convert the ON condition from sqlparser Expr to datafusion Expr
+ let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
+
+ // 5. Convert each WHEN clause
+ let df_clauses = clauses
+ .into_iter()
+ .map(|clause| {
+ self.merge_clause_to_plan(
+ clause,
+ &combined_schema,
+ &target_schema,
+ &target_qualifier,
+ &mut planner_context,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+
+ // 6. Build the MERGE operation. Column references to the target may be
+ // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`).
+ // Canonicalize those to the real target table qualifier so the stored
+ // plan is independent of the alias: this lets the analyzer passes and
+ // proto deserialization rebuild the target schema from `table_name`
+ // alone, without carrying the alias as extra state.
+ let mut merge_op = MergeIntoOp {
+ on: on_expr,
+ clauses: df_clauses,
+ };
+ if target_qualifier != target_table_ref {
+ // Target references in correlated subqueries are represented as
+ // `OuterReferenceColumn`s inside the embedded logical plan. The
+ // alias canonicalization below only rewrites top-level expression
+ // columns, so accepting such a subquery would leave the target
+ // alias in the public MERGE representation. Reject this case until
+ // the alias can be rewritten scope-safely inside subquery plans.
+ for expr in merge_op.exprs() {
+ if Self::has_outer_reference_to_qualifier(expr, &target_qualifier)? {
+ return not_impl_err!(
+ "MERGE subqueries correlated to target alias \
+ '{target_qualifier}' are not supported"
+ );
+ }
+ }
+
+ // Canonicalizing target columns to `target_table_ref` is only safe
+ // when the source does not already use that qualifier. If it does
+ // (e.g. `MERGE INTO target AS t USING source AS target`), the two
+ // namespaces would collapse and later resolution could silently
+ // pick the source column for a target reference. Reject that
+ // collision rather than change the meaning of the condition.
+ if source_plan.schema().iter().any(|(qualifier, _)| {
+ qualifier.is_some_and(|q| q.resolved_eq(&target_table_ref))
+ }) {
+ return plan_err!(
+ "MERGE source may not use the target table name '{target_table_ref}' \
+ as a qualifier while the target is aliased as '{target_qualifier}'; \
+ use a different source alias"
+ );
+ }
+ let canonical = merge_op
+ .exprs()
+ .into_iter()
+ .cloned()
+ .map(|expr| {
+ Self::canonicalize_target_qualifier(
+ expr,
+ &target_qualifier,
+ &target_table_ref,
+ )
+ })
+ .collect::<Result<Vec<_>>>()?;
+ merge_op = merge_op.with_new_exprs(canonical)?;
+ }
+
+ Ok(LogicalPlan::Dml(DmlStatement::new(
+ target_table_ref,
+ target_table_source,
+ WriteOp::MergeInto(Box::new(merge_op)),
+ Arc::new(source_plan),
+ )))
+ }
+
+ /// Rewrite every [`Expr::Column`] qualified with `from` to instead use
+ /// `to`, leaving all other columns untouched. Used to canonicalize MERGE
+ /// target-alias references to the real target table qualifier.
+ fn canonicalize_target_qualifier(
+ expr: Expr,
+ from: &TableReference,
+ to: &TableReference,
+ ) -> Result<Expr> {
+ expr.transform(|expr| match expr {
+ Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
+ Transformed::yes(Expr::Column(Column::new(Some(to.clone()), col.name))),
+ ),
+ other => Ok(Transformed::no(other)),
+ })
+ .map(|transformed| transformed.data)
+ }
+
+ /// Return true if an expression contains a subquery whose embedded plan
+ /// has an outer reference qualified by `qualifier`.
+ fn has_outer_reference_to_qualifier(
+ expr: &Expr,
+ qualifier: &TableReference,
+ ) -> Result<bool> {
+ let mut found = false;
+ expr.apply(|expr| {
+ let subquery = match expr {
+ Expr::Exists(exists) => Some(&exists.subquery),
+ Expr::InSubquery(in_subquery) => Some(&in_subquery.subquery),
+ Expr::SetComparison(set_comparison) => Some(&set_comparison.subquery),
+ Expr::ScalarSubquery(subquery) => Some(subquery),
+ _ => None,
+ };
+
+ if let Some(subquery) = subquery {
+ subquery.subquery.apply_with_subqueries(|plan| {
+ plan.apply_expressions(|expr| {
+ expr.apply(|expr| {
+ if let Expr::OuterReferenceColumn(_, column) = expr
+ && column.relation.as_ref() == Some(qualifier)
Review Comment:
Thanks for catching this. I implemented a scope-aware guard in the stacked follow-up PR https://github.com/wirybeaver/datafusion/pull/4. It now allows nested aliases to shadow the MERGE target alias while continuing to reject genuine direct and deeply nested target correlation. I also added consolidated regressions covering projection scopes, multi-level nesting, lateral joins, and LIMIT/OFFSET scope behavior. The clean stacked PR is in my fork because the parent `feature/mergeinto` branch exists only there; it can be retargeted to `apache/datafusion:main` after #22988 lands.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO [datafusion]
Posted by "kosiew (via GitHub)" <gi...@apache.org>.
kosiew commented on PR #22988:
URL: https://github.com/apache/datafusion/pull/22988#issuecomment-5068475821
@wirybeaver
Can you resolve the merge conflicts?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org