feat: add basic MemTable MERGE INTO support

[wirybeaver]

Which issue does this PR close?

Follow-up to the MERGE INTO review in #22988. No separate issue.

Rationale for this change

MERGE INTO can be planned, but the built-in in-memory table provider cannot execute it. This makes the SQL surface difficult to exercise end-to-end with DataFusion's default provider.

What changes are included in this PR?

• Implements basic MemTable MERGE INTO execution for matched update/delete, not-matched insert, and not-matched-by-source update/delete.
• Defers mutation until physical-plan execution so planning and EXPLAIN remain side-effect free.
• Handles first matching clause semantics, NULL predicates as false, duplicate source matches for a target row as an error before mutation, insert column subsets with defaults or typed NULLs, expression casts to target column types, affected-row counts, atomic target replacement, partition preservation, and sort-order reset.
• Documents MERGE INTO syntax, provider caveats, and currently unsupported planner syntax.
• Extends sqllogictests and Rust tests for execution, NULL/no-op predicates, defaults, duplicate-match errors, and multi-partition output.

The logical-plan display changes were split into independent PR #26038.

What is the testing strategy for this PR?

• cargo fmt --all
• cargo check -p datafusion-catalog -p datafusion
• cargo test -p datafusion-catalog
• cargo test -p datafusion merge_into_ -- --nocapture
• cargo test -p datafusion --test core_integration merge_into -- --nocapture
• cargo test --profile ci --test sqllogictests -- merge_into.slt --nocapture
• ./ci/scripts/doc_prettier_check.sh --write --allow-dirty
• cargo clippy --all-targets --all-features -- -D warnings
• Full extended workspace test command from the contributor guide

Are there any user-facing changes?

Yes. MemTable now supports basic MERGE INTO and returns the number of affected rows. There are no public API signature changes.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 80.59259% with 131 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.65%. Comparing base (c3ef346) to head (2b71a51).

Files with missing lines	Patch %	Lines
datafusion/catalog/src/memory/table.rs	80.10%	78 Missing and 38 partials ⚠️
datafusion/core/src/datasource/memory_test.rs	83.69%	5 Missing and 10 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24195      +/-   ##
==========================================
- Coverage   82.66%   82.65%   -0.01%     
==========================================
  Files        1147     1147              
  Lines      446357   447033     +676     
  Branches   446357   447033     +676     
==========================================
+ Hits       368971   369494     +523     
- Misses      54997    55100     +103     
- Partials    22389    22439      +50     

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

[timsaucer]

It looks like this will mutate the table even when we don't execute. If you run these commands in datafusion-cli you'll see:

CREATE TABLE destination(id INT, val VARCHAR);
INSERT INTO destination VALUES (1, 'a'), (2, 'b'), (3, 'c');
CREATE TABLE source(id INT, val VARCHAR);
INSERT INTO source VALUES (2, 'B');
SELECT * FROM destination ORDER BY id;
EXPLAIN MERGE INTO destination USING source ON destination.id = source.id
WHEN MATCHED THEN UPDATE SET val = source.val;
SELECT * FROM destination ORDER BY id;

Yields this output:

0 row(s) fetched.
Elapsed 0.024 seconds.

+-------+
| count |
+-------+
| 3     |
+-------+
1 row(s) fetched.
Elapsed 0.018 seconds.

0 row(s) fetched.
Elapsed 0.000 seconds.

+-------+
| count |
+-------+
| 1     |
+-------+
1 row(s) fetched.
Elapsed 0.000 seconds.

+----+-----+
| id | val |
+----+-----+
| 1  | a   |
| 2  | b   |
| 3  | c   |
+----+-----+
3 row(s) fetched.
Elapsed 0.006 seconds.

+---------------+-------------------------------+
| plan_type     | plan                          |
+---------------+-------------------------------+
| physical_plan | ┌───────────────────────────┐ |
|               | │      CooperativeExec      │ |
|               | │    --------------------   │ |
|               | │      CooperativeExec      │ |
|               | └─────────────┬─────────────┘ |
|               | ┌─────────────┴─────────────┐ |
|               | │       DmlResultExec       │ |
|               | │    --------------------   │ |
|               | │       DmlResultExec:      │ |
|               | │        rows_affected      │ |
|               | │             :             │ |
|               | │             1             │ |
|               | └───────────────────────────┘ |
|               |                               |
+---------------+-------------------------------+
1 row(s) fetched.
Elapsed 0.003 seconds.

+----+-----+
| id | val |
+----+-----+
| 1  | a   |
| 2  | B   |
| 3  | c   |
+----+-----+
3 row(s) fetched.
Elapsed 0.000 seconds.

The big difference I want to point out is in the two select commands. In the first one we have values a, b, c and in the second we have a, B, c so the table was mutated even though we only did an explain plan. I didn't expect explain to actually do the mutation.

Now I think this is a pre-existing issue. In fact if you run an insert command you have similar problems. Starting in a fresh cli:

INSERT INTO t2 VALUES (1), (2);
SELECT * FROM t2;
EXPLAIN UPDATE t2 SET a = 99;
SELECT * FROM t2;

Yields:

0 row(s) fetched.
Elapsed 0.024 seconds.

+-------+
| count |
+-------+
| 2     |
+-------+
1 row(s) fetched.
Elapsed 0.018 seconds.

+---+
| a |
+---+
| 1 |
| 2 |
+---+
2 row(s) fetched.
Elapsed 0.002 seconds.

+---------------+-------------------------------+
| plan_type     | plan                          |
+---------------+-------------------------------+
| physical_plan | ┌───────────────────────────┐ |
|               | │      CooperativeExec      │ |
|               | │    --------------------   │ |
|               | │      CooperativeExec      │ |
|               | └─────────────┬─────────────┘ |
|               | ┌─────────────┴─────────────┐ |
|               | │       DmlResultExec       │ |
|               | │    --------------------   │ |
|               | │       DmlResultExec:      │ |
|               | │        rows_affected      │ |
|               | │             :             │ |
|               | │             2             │ |
|               | └───────────────────────────┘ |
|               |                               |
+---------------+-------------------------------+
1 row(s) fetched.
Elapsed 0.003 seconds.

+----+
| a  |
+----+
| 99 |
| 99 |
+----+
2 row(s) fetched.
Elapsed 0.000 seconds.

So again we're getting mutation of the data during an explain. I'm not sure if we want to merge as is, make a change, or open an issue. Do others have thoughts? Is this a well known and accepted approach - mutation on explain?

[wirybeaver]

@timsaucer Thanks for catching this. Fixed in e2e3647: MemTable now returns a deferred MergeIntoExec, so source collection and table mutation happen only when the physical plan is executed. Plain EXPLAIN MERGE INTO now builds and displays the plan without mutating the target; EXPLAIN ANALYZE still executes it as expected.

I also added a regression test that verifies the target rows remain unchanged after EXPLAIN MERGE INTO. The pre-existing eager UPDATE/DELETE behavior is separate from this PR.

[martin-g]

Now I think this is a pre-existing issue. In fact if you run an insert command you have similar problems

This has been addressed with #25040 for UPDATE and DELETE.

[martin-g]

This will be rather slow for bigger tables - O(n*m) + a new RecordBatch allocation for every row.

[wirybeaver]

Partially addressed in 062af76. Generic ON predicates still require considering target/source pairs, but evaluation now runs in bounded 8K-pair Arrow batches instead of allocating one RecordBatch per pair. I am leaving this thread open because the generic algorithm remains O(n*m).