feat(python): preserve re-scannable merge sources

[easyrev]

Problem

Python currently converts LanceDataset and Arrow dataset inputs into RecordBatchReaders before passing them to Rust. This loses the information needed to distinguish re-scannable sources from one-shot streams.

With conflict retries enabled, merge_insert buffers these inputs for replay, even when the original source can be scanned again. The buffer consumes additional memory and may spill to disk, regardless of whether a conflict occurs.

This PR preserves source repeatability across the Python/Rust boundary, following the direction discussed in #7895 and #7368.

Implementation

Source registry

Replace the separate source classification and conversion logic with a singledispatch registry.

Each adapter returns a WriteSource containing the schema, reader factory, and one of three modes: materialized, replayable, or streaming.

Third-party types can register their own adapters through lance.types.coerce_source.

Re-scannable provider

Add a Rust TableProvider that creates a fresh reader for each scan and uses the existing provider-based merge retry path.

The provider invokes Python reader factories on blocking threads to avoid blocking Tokio workers. It also checks each reader's schema, including metadata.

Source eligibility

Automatically use the re-scannable path for:

• Native Lance scans with stable settings and a pinned dataset version.
• Unfiltered built-in Arrow datasets, including supported in-memory, filesystem, and union datasets.

Repeated scans must preserve the same schema, values, and row order. Arrow source files must remain unchanged during the operation.

Opaque Arrow scanners and sources with uncertain values or ordering retain buffered replay. This includes scanners that may consume their input only once. Callers with a stronger repeatability guarantee can provide an explicit WriteSource.

Existing materialized and one-shot input behavior remains supported. Uncommitted execution still consumes its source once without committing.

Plan analysis

Route analyze_plan through the same source strategy as merge execution.

Also fix the shared analysis helper to propagate execution errors and use the correct output schema. Previously, a source failure could produce a misleading successful analysis result.

Validation

Correctness

• Real conflict retries across nine source forms, including a competing upsert that changes the target's matching rows.
• Repeated reader creation, source version pinning, scan settings, row ordering, and schema metadata.
• Streaming fallback, factory failures, reader cleanup, and error propagation.
• Single-Tokio-worker retry execution.
• Compatibility checks with PyArrow 14.0.2 and 25.0.1.

Tests and checks

• 106 focused Python tests passed.
• 137 additional write regression tests passed.
• 4 Rust provider tests, 1 source API doctest, and the shared analysis helper test passed.
• Python lint and Rust Clippy checks passed.

The full Python and Rust test suites and remote CI have not run yet.

Performance

Compared the upstream implementation and this change using matched release-with-debug builds.

The benchmark covers 1.125 MiB and 132 MiB inputs, three source types, and both conflict and no-conflict cases. Each configuration was measured five times, for 120 primary measurements.

Replay avoidance was verified. Eligible sources bypassed replay-buffer initialization in all measured cases. One-shot inputs and opaque scanners retained their existing behavior.

For 132 MiB inputs, replay spill decreased from approximately 132 MiB to zero.

Without a conflict, median merge latency decreased:

Source	Upstream	This PR	Change
LanceDataset	230.82 ms	218.17 ms	−5.5%
Arrow Dataset	201.76 ms	193.26 ms	−4.2%

The performance tradeoff depends on the workload:

• A separate probe-free test found small Lance merges approximately 0.75 ms slower at the median, with overlapping measurement ranges.
• With a real retry, the 132 MiB Lance case used 6.8% more total CPU time and approximately 39 MiB more peak process RSS.
• Conflict wall time includes randomized retry backoff, which limits latency comparisons.

These measurements establish replay-buffer avoidance for the tested sources. They do not establish a general latency or memory improvement.

The results cover local warm-cache workloads. Remote storage and cold-cache performance have not been measured.

Public API

Expose WriteSource and coerce_source through lance.types as experimental APIs, without top-level re-exports.

This allows third-party source adapters while leaving room to refine the reader-factory contract.

Feedback on the public API names, location, and experimental status is welcome.

Closes #7895

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.