fix(merge): partition pruning, EXPLAIN routing, and data-loss fix on the MERGE path

[rampage644]

Four related Embucket MERGE-path fixes, all needed to make MERGE INTO correct, observable, and tunable on partitioned Iceberg targets.

Changes

1. fix(custom_type_coercion) — CustomTypeCoercionRewriter::analyze_internal now falls back to plan.schema() when plan.inputs() is empty (leaf TableScan case). Without this, the target filter that merge_query injects via LogicalPlanBuilder::scan_with_filters against a partitioned DataFusionTable failed the analyzer with Schema error: No field named <col>. Unblocks source-side partition-hint pruning.

2. feat(merge): EXPLAIN routing — Embucket has its own MERGE planner, so DataFusion rejected EXPLAIN MERGE INTO … with Unsupported SQL statement. Split merge_query into a pure plan-builder + a thin executing wrapper; in execute(), when the parsed statement is Explain(MergeInto), build the plan and wrap it in LogicalPlan::Explain / LogicalPlan::Analyze to match DataFusion's explain_to_plan. Adds an EXPLAIN MERGE snapshot test (22 existing merge_into tests stay green).

3. feat(merge): MetricsSet — surface updated_rows / inserted_rows / deleted_rows as per-clause counters on MergeIntoCOWSinkExec so EXPLAIN ANALYZE MERGE INTO … now shows them alongside child DataSourceExec scan metrics (bytes_scanned, files_ranges_pruned_statistics, row_groups_pruned_statistics, pushdown_rows_pruned, output_rows, elapsed_compute).

4. fix(merge): preserve target rows — fixes MERGE INTO silently drops unmatched target rows when target is unsorted by the ON-key #128. A stale fast-path guard in MergeCOWFilterStream::poll_next short-circuited on matching_data_and_manifest_files.is_empty() alone, without checking all_matching_data_files. So when a target file had been seen as matching in an earlier batch and a later batch contained only target rows for that file, the rows in the later batch were silently dropped (the COW commit then overwrote the original file with the partial result). Six-line guard tightening: also require all_matching_data_files to be empty before short-circuiting. Counters were always correct; only the committed parquet was short. Sorting either input by the join key masked the bug, which is why it took finding to surface.

Why all four together

(1) and (2)+(3) work in tandem: with (1) alone the source-side filter is pushed into the target TableScan as a partial_filter and the MERGE commits the right rows, but the prune is invisible from outside. (2)+(3) make EXPLAIN MERGE and EXPLAIN ANALYZE MERGE work end-to-end and surface the prune metrics, so an operator can verify and tune the pruning. (4) is the silent-data-loss bug those tools exposed once the bigger MERGE workloads started running on real partitioned Iceberg tables.

Tests

• 3 new unit tests on MergeCOWFilterStream covering the matching-then-target-only patterns (fail on main, pass here)
• 1 new SQL snapshot test merge_into_mixed_unsorted_multi_row_no_data_loss (10-row target × 4-row source, asserts no row loss)
• 1 new SQL snapshot test for EXPLAIN MERGE (EXPLAIN ANALYZE is exercised end-to-end against a Lambda but intentionally not snapshot-tested — its formatted-table widths vary per run)
• cargo test -p executor: 362 passed, 0 failed

End-to-end validation against deployed Embucket Lambda

#128 isolated repro — 5 consecutive runs against a freshly-deployed lambda built from this branch, all loss=0 deterministic:

target=2101659 source=2103638 merge=[1052088, 1051550] post=3153747 expected=3153747 loss=0
target=2101659 source=2103638 merge=[1052088, 1051550] post=3153747 expected=3153747 loss=0
target=2101659 source=2103638 merge=[1052088, 1051550] post=3153747 expected=3153747 loss=0
target=2101659 source=2103638 merge=[1052088, 1051550] post=3153747 expected=3153747 loss=0
target=2101659 source=2103638 merge=[1052088, 1051550] post=3153747 expected=3153747 loss=0

12.6 M target × 6.3 M source: post=14697755 expected=14697755 loss=0.

dbt-snowplow-web stretch validation — fresh seed + R1–R5 of loop_dbt.py, complete. user_mapping Δ is the strongest MERGE-correctness signal (its source CTE feeds straight from the buggy MERGE on lifecycle_manifest):

Round	dbt	Δ user_mapping (this PR)	Δ user_mapping (pre-fix)
1	280.9 s	+1,702,223	+1,702,223
2	381.1 s	+1,699,998	+1,699,998
3	386.0 s	+1,700,804 ✓	+690,664 ❌
4	390.3 s	+1,699,815 ✓	+186,950 ❌
5	385.5 s	+1,699,577 ✓	+611,373 ❌
total after R5		10,204,627	6,593,418

R3 is where the bug first manifested in the original buggy run. With the fix, R3, R4, and R5 each deliver the full ~+1.7 M rows that match the day-partitioned baseline. The fix recovered 3,611,209 rows of user_mapping data (≈55% of the post-fix total) that the buggy MERGE was silently dropping over 5 rounds.

Source-side partition pruning end-to-end: MERGE INTO atomic.events_hooli_tiny USING atomic.events_hooli_ident ON t.event_id = s.event_id WHEN MATCHED THEN UPDATE … — previously failed with custom_type_coercion / Schema error: No field named event_name, now returns 100 matched rows and commits. The companion EXPLAIN ANALYZE MERGE INTO … emits the target TableScan's partial_filters=[event_name = Utf8("ad_start_event")] and full DataSourceExec scan-metrics block.

Out of scope (separate follow-ups)

• MergeCOWFilterStream::not_matched_buffer LRU eviction — capacity is 2 on Lambda; doesn't fire in single-target-file repros but will silently evict any third-and-later concurrently-buffered file's rows. Worth its own issue with a 3+ data file repro.
• Round-6 dbt SELECT-phase OOM — independent of this PR. With the data-loss fix in place, lifecycle_manifest will now grow correctly and the model's previous_sessions LEFT JOIN OOMs the 9 GB Lambda pool around R6. Needs planner-level predicate pushdown or non-Lambda compute.

Companion to Embucket/iceberg-rust#57. Closes #128.