feat: implement support for INSERT ... ON CONFLICT (DO NOTHING / DO UPDATE) clauses

[Nachiket-Roy]

Which issue does this PR close?

• Closes [EPIC] Complete DML Support (MERGE, INSERT OVERWRITE, TRUNCATE) #19617

Rationale for this change

DataFusion previously rejected PostgreSQL-style INSERT INTO ... ON CONFLICT (col, ...) DO NOTHING and INSERT INTO ... ON CONFLICT (col, ...) DO UPDATE SET ... [WHERE ...] upsert statements during SQL planning with "This feature is not implemented: ON CONFLICT is not supported".

Design & Architectural Approach

Rather than modifying InsertOp or adding new dispatch methods on TableProvider (which would break external connectors such as delta-rs, iceberg-rust, and custom catalogs across upgrades) or adding protobuf schema churn, this change desugars INSERT ... ON CONFLICT into WriteOp::MergeInto(Box<MergeIntoOp>) at the SQL planning stage:

1. 0% Breaking Change: InsertOp remains unchanged (Append, Overwrite, Replace), and TableProvider::insert_into is untouched.
2. 0% Protobuf Churn: Reuses the existing dml_node::Type::MergeInto protobuf representation.
3. Instant Connector Support: Any lakehouse connector or catalog implementing TableProvider::merge_into automatically gains ON CONFLICT upsert capabilities for free.
4. PostgreSQL Semantics:
   • The incoming dataset is aliased to the standard PostgreSQL pseudo-relation excluded.
   • DO NOTHING maps to WHEN NOT MATCHED THEN INSERT.
   • DO UPDATE maps to WHEN MATCHED [AND predicate] THEN UPDATE SET ... followed by WHEN NOT MATCHED THEN INSERT.
   • NULL conflict keys never conflict (NULL != NULL), taking the insert path.
   • Intra-batch hazards: duplicate conflict keys within a single statement produce "ON CONFLICT DO UPDATE command cannot affect row a second time", while DO NOTHING coalesces/deduplicates them.
   • Unconstrained tables (constraints() == None or empty): validated against existing schema columns, and guarded against duplicate rows at runtime.

What changes are included in this PR?

1. SQL Planning & Desugaring (datafusion-sql)

• Modified datafusion/sql/src/statement.rs:
  • Removed unconditional rejection of the on conflict clause.
  • Enforced mutual exclusivity: ON CONFLICT cannot be combined with INSERT OVERWRITE or REPLACE INTO.
  • Reserved alias check: Rejects target table names/aliases named excluded.
  • Wrapped the incoming query in SubqueryAlias("excluded").
  • Verified conflict columns exist in the table schema.
  • Constraint validation fallback: When table_source.constraints() contains defined constraints (!constraints.is_empty()), strictly validates that target columns match a Constraint::PrimaryKey or Constraint::Unique (failing with "There is no unique or exclusion constraint matching the ON CONFLICT specification" if mismatched). When constraints are empty or None, allows the operation so real connectors are not blocked.
  • Constructed MergeIntoOp with equality join conditions between target columns and excluded columns.

2. MemTable Reference Execution (datafusion-catalog)

• Implemented TableProvider::merge_into on MemTable in datafusion/catalog/src/memory/table.rs:
  • Deadlock-Free Locking: Locks all partitions in strict ascending index order (0..N-1).
  • Cross-Partition Equi-Join Index: Builds HashMap<RowKey, (partition_idx, batch_idx, row_idx)>.
  • Runtime Duplicate Key Detection: Aborts with an explicit error if duplicate keys are detected in unconstrained tables during index construction.
  • Intra-Batch Hazard Prevention: Aborts on duplicate keys for DO UPDATE and coalesces duplicates for DO NOTHING.
  • NULL Conflict Key Handling: Rows containing NULL in any conflict column bypass conflict detection and take the insert path.
  • Robust NOT MATCHED Projection: Evaluates insert expressions against a combined not-matched row batch, correctly filling unmentioned target columns with typed nulls and applying type casts where required.
  • Unaliased ON Expression Parser: Added extract_column to unwrap nested Expr::Alias and Expr::Cast when extracting equi-join keys.
  • Batch Updates & Deletions Rebuild: Updates modified columns, removes deleted rows, appends new rows, and emits affected row counts via DmlResultExec.

What is the testing strategy for this PR?

1. End-to-End Sqllogictests:

   • Added datafusion/sqllogictest/test_files/insert_on_conflict.slt covering:
     • Basic insert and DO NOTHING row skipping.
     • DO UPDATE modifications with and without WHERE predicates (where excluded.score > users.score).
     • NULL conflict keys bypassing conflict detection (NULL != NULL).
     • Intra-batch duplicate keys failing on DO UPDATE and coalescing on DO NOTHING.
     • Multi-column composite conflict keys (a, b).
     • Runtime duplicate key detection on unconstrained tables.
     • Planning errors (mutual exclusivity with overwrite, reserved alias excluded, non-existent columns, duplicate assignments).
   • Updated datafusion/sqllogictest/test_files/merge_into.slt to verify physical plan execution on MemTable.

2. Unit & Integration Tests:

   • datafusion/sql/tests/sql_integration.rs: Added integration tests verifying planned MergeIntoOp structures, predicates, and aliases for DO NOTHING and DO UPDATE.
   • datafusion/sql/tests/sql_integration.rs & datafusion/sql/tests/common/mod.rs: Added 6 negative test cases in test_insert_schema_errors checking mutual exclusivity, non-existent columns, duplicate assignments, and constraint mismatches.

3. Format & Lint:

   • Passed cargo fmt --all -- --check.
   • Passed cargo clippy -p datafusion-sql -p datafusion-catalog --all-targets --all-features -- -D warnings with zero warnings.

Are there any user-facing changes?

• Users can now execute PostgreSQL-style INSERT INTO ... ON CONFLICT (...) DO NOTHING and INSERT INTO ... ON CONFLICT (...) DO UPDATE SET ... statements in SQL.
• MemTable now supports executing MERGE INTO queries in-memory.
• No breaking API changes: InsertOp is completely unchanged, and existing connector implementations remain 100% compatible.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.90937% with 133 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.90%. Comparing base (4cb8c21) to head (9faa820).
⚠️ Report is 71 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/catalog/src/memory/table.rs	77.54%	80 Missing and 26 partials ⚠️
datafusion/sql/src/statement.rs	85.78%	20 Missing and 7 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24982      +/-   ##
==========================================
+ Coverage   81.72%   81.90%   +0.18%     
==========================================
  Files        1127     1132       +5     
  Lines      416310   421969    +5659     
  Branches   416310   421969    +5659     
==========================================
+ Hits       340219   345625    +5406     
+ Misses      56100    55901     -199     
- Partials    19991    20443     +452     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[quwin]

Thanks @Nachiket-Roy!
I've found what I think is an execution bug while reviewing, where WHEN NOT MATCHED BY SOURCE AND <predicate> THEN DELETE ignores the predicate and deletes every unmatched target row.

In datafusion/catalog/src/memory/table.rs, the loop evaluates clause.predicate only in the MergeIntoAction::Update arm.

The MergeIntoAction::Delete arm pushes the row into row_deletions and increments affected_count without checking the predicate.

That means that the rows the user asked to keep (predicate false or NULL) are deleted.

Repro (run as a sqllogictest case):

CREATE TABLE t_repro(id int, flag boolean);
INSERT INTO t_repro VALUES (1, true), (2, false), (3, NULL);
CREATE TABLE s_repro(id int);
INSERT INTO s_repro VALUES (2);
MERGE INTO t_repro USING s_repro ON t_repro.id = s_repro.id
  WHEN MATCHED THEN UPDATE SET flag = false
  WHEN NOT MATCHED BY SOURCE AND t_repro.flag THEN DELETE;
SELECT id, flag FROM t_repro ORDER BY id;

Expected: 2 false and 3 NULL.
Actual: only 2 false.

Suggested fix: evaluate clause.predicate in the MergeIntoAction::Delete arm and add an execution test for the predicate-false and predicate-NULL cases.
The current merge_into.slt coverage for this clause is explain-only.

[Nachiket-Roy]

Hello @quwin,
Thanks for the review. I have fixed the WHEN NOT MATCHED BY SOURCE Predicate in DELETE. Please take a look again!