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 75.30675% with 161 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.65%. Comparing base (722cbf2) to head (e3c90f3).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/catalog/src/memory/table.rs	70.68%	114 Missing and 22 partials ⚠️
datafusion/sql/src/statement.rs	86.70%	19 Missing and 6 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24982      +/-   ##
==========================================
- Coverage   81.67%   81.65%   -0.02%     
==========================================
  Files        1126     1126              
  Lines      414533   415171     +638     
  Branches   414533   415171     +638     
==========================================
+ Hits       338562   339025     +463     
- Misses      56058    56203     +145     
- Partials    19913    19943      +30     

☔ 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.