fix(merge_insert): make analyze_plan follow execute's source routing

[LuciferYang]

What this fixes

MergeInsertBuilder.analyze_plan(data) coerced every input to a one-shot stream, so it reported the streaming plan even when execute(data) on the same input would run a different one (#8771).

Which side of the hash join gets collected, and the join type, both follow the statistics the source reports. execute sends a materialized source through an in-memory table that reports an exact row count and byte size, and DataFusion's JoinSelection picks the collected side from that. A stream reports nothing. analyze_plan reported the wrapping it chose rather than the one execute would choose, so anyone profiling a merge read metrics off the wrong side of the join.

analyze_plan now dispatches on _is_materialized exactly as execute does.

What the diagnostic printed, and what it prints now

The docstring example in dataset.py passes a pa.table. Before, for that input:

      HashJoinExec: mode=CollectLeft, join_type=Right, ...
        LanceRead: ...
        RepartitionExec: ...
          ProjectionExec: expr=[..., true as __merge_source_sentinel]
            StreamingTableExec: ...

After:

      RepartitionExec: ...
        HashJoinExec: mode=CollectLeft, join_type=Left, ...
          ProjectionExec: expr=[..., true as __merge_source_sentinel]
            DataSourceExec: ...
          LanceRead: ...

The second one is what execute has been running all along. The doctest asserted the first.

Rust surface

MergeInsertJob gains analyze_plan_batches and analyze_plan_provider, mirroring the existing execute_batches and execute_provider. analyze_plan(stream) keeps its signature and delegates to the provider entry, so external Rust callers still compile and a stream is still reported as a stream.

Two doc corrections came out of reviewing this. explain_plan now says outright that it only ever reports the streaming shape, because it receives a schema rather than data and so cannot know how the source would be wrapped; it also points at analyze_plan while noting that analyze_plan runs the merge and may write data files, which explain_plan does not. And analyze_plan_batches documents the two cases where it reports the streaming shape anyway: SourceDedupeBehavior::FirstSeen re-wraps the source in a stream ahead of the join, and an empty batch list carries no schema so the provider falls back to the dataset's.

What this does not change

No execution behaviour. execute already routed materialized sources through the in-memory table; only the diagnostic was out of step with it.

A materialized analyze_plan now collects the reader into memory in Rust before planning, where it used to stream. The inputs _is_materialized accepts are already fully in memory, so the extra copy is bounded by data the caller holds, and it is the same copy execute has always made.

The source types that could report statistics but do not are untouched. lance.LanceDataset, pa.dataset.Dataset, and pa.dataset.Scanner all arrive as a bare reader through _coerce_reader even though each knows its row count and can be scanned again, and the default streaming path drains the whole source into a spill before reporting no statistics at all. Both are remaining bullets on #4583, and this change is what makes their effect visible from Python.

One pre-existing gap this touches without fixing: batches_to_provider falls back to the dataset's schema when the batch list is empty, so a zero-batch materialized source is validated against the target's columns rather than its own. execute_batches and execute_uncommitted_batches have always done this, and closing it changes execute's public behaviour from a silent no-op to an error, which needs its own change and its own tests.

One drive-by, disclosed rather than hidden: explain_plan's not-supported message said only full-schema sources are supported. can_use_create_plan accepts a subset schema and, for a delete-only merge, the join keys alone, and its own doc comment lists all three. Rewriting the sibling message on the analyze_plan path made the two contradict each other, so both now name the two real reasons instead. The does not support explain_plan prefix that four tests match on is unchanged.

Test plan

• New test_merge_insert_analyze_plan_matches_execute_routing: a pa.Table source must report DataSourceExec and join_type=Left, a RecordBatchReader must report StreamingTableExec and join_type=Right. The first assertion fails without the dispatch change.
• New test_analyze_plan_reports_the_given_source_shape covers the three Rust entries, including analyze_plan_provider directly.
• New test_plan_join_build_side_follows_source_statistics pins which side the join collects at both of DataFusion's decision points: past hash_join_single_partition_threshold_rows where only the source can be collected, and below it where the smaller side wins.
• cargo test -p lance --lib merge_insert -- --test-threads=1: 220 pass.
• uv run pytest python/tests/test_dataset.py -k merge_insert: 26 pass.
• uv run pytest --doctest-modules python/lance/dataset.py -k "explain_plan or analyze_plan": 2 pass.
• cargo fmt --all, cargo clippy --all --tests --benches -- -D warnings, uv run make lint from python/.

[lance-gatekeeper]

✅ Gate recommendation: approve.

The latest merge from main preserves the feature patch. Materialized analysis still follows the statistics-bearing provider route used by execution, streams remain streaming, and the routing and join-selection regressions pass against the updated base.

[wjones127]

Looks good. Thanks for the fix!

[LuciferYang]

Thank you @wjones127