perf(merge_insert): probe the most selective indexed key first and stop early

[LuciferYang]

What this changes

MapIndexExec::map_batch probed every indexed join key as one ScalarIndexExpr::And and evaluated the whole expression, so a low-cardinality key materialized a candidate set the size of the table while pruning almost nothing (#8718).

It now probes one key at a time, intersects as it goes, and stops once the candidate set is no larger than the source batch. Keys are ordered by how many distinct values the source batch holds for them, so a caller who writes a composite key coarse-to-fine (["tenant_id", "row_id"]) does not pay for the coarse probe first. IsIn lists are deduplicated. Both the ordering and the dedup are gated on having more than one lookup, so the single-key path behaves as before.

Skipping a probe widens the candidate set, which is safe because the downstream join filters on the full composite key; the file already documents that the probe result is a super-set. Extra candidates stay inside index-covered fragments because the restricted deletion mask is still built from every lookup's fragment bitmap. That coupling is now stated in a comment, because relaxing it would feed the same target row into the join twice and the default SourceDedupeBehavior::Fail would abort the merge.

Numbers

Same dataset as the issue. Baseline and patched were measured in one sitting about twenty minutes apart, swapping only the binary: the patch was reverse-applied for the baseline, then re-applied. Release build, warm index cache, median of 5 runs.

case	main	this PR
on = ["composite_a", "composite_b"]	198.9 ms	10.2 ms
on = ["composite_b", "composite_a"]	199.4 ms	10.5 ms
on = ["composite_a"], single key	10.2 ms	9.8 ms
conditional_update, single key	11.1 ms	10.1 ms
id_int, single key	10.9 ms	9.9 ms
id_uuid4 100k, single key	1132.5 ms	1114.5 ms
composite v2 hash path (control)	128.7 ms	124.3 ms
id_uuid4 v2 hash path (control)	228.6 ms	221.3 ms

The two controls do not touch MapIndexExec and moved by 3.2% and 3.4%, so the two sides are comparable. The single-key shapes moved by 1.6% to 9.2%, which lands inside this script's run-to-run spread once the control drift is subtracted; that path is meant to be unchanged.

What this does not fix

A single low-cardinality key still materializes the whole candidate set. There is no second key to intersect with and no reason for the loop to stop, so on = ["composite_b"] alone still exhausts the default pool, before and after this change. Bounding it needs a budget inside the index probe, and merge_insert would first need to accept a non-exact result, which it currently todo!()s. Related: #1983.

The distinct count is a source-side proxy for target-side selectivity. A skewed batch, with many distinct values that each match many target rows, can be ordered worse than the caller wrote it; the floor is the old behaviour of probing every key.

Probes now run in sequence where the old And ran them concurrently through try_join!. That is what makes stopping possible. Measured on the case where no probe can be skipped, a 100-row source with on = ["composite_b", "composite_a"] so that the coarse key probes first and returns roughly 976k candidates: 36.6 ms to 43.4 ms on a cold index cache, 10.6 ms to 11.6 ms warm, so 19% and 9%. The same construction with the columns swapped stops after one probe and goes from 37.1 ms to 3.5 ms cold.

Test plan

• New map_index_exec_probes_most_selective_key_first asserts the emitted candidate count, which is the only observable that reveals which probe ran: IndexMetrics has no per-probe counter and one collector is shared across lookups. Reverting the ordering makes it fail with left: 2, right: 4.
• cargo test -p lance --lib merge_insert -- --test-threads=1: 219 pass.
• cargo fmt --all, cargo clippy --all --tests --benches -- -D warnings.
• Benchmark numbers above.

Follow-up note

The route gate comment at merge_insert.rs:1432 says a partially-indexed composite key "under-matches" on the indexed path. The mechanism is the opposite: each column's IsIn is a super-set, uncovered fragments go through the union scan, and the full-key join trims. Left alone to keep this diff to the two files it needs.

Closes #8718

[LuciferYang]

Benchmarked the sequential-probe path, since that was the open item.

The case is constructible on the existing merge_insert_narrow dataset (10M rows, composite_a unique, composite_b = row_index % 1024, BTree on both). A 100-row source with on = ["composite_b", "composite_a"] gives both keys 100 distinct source values, so the ordering ties, the stable sort keeps the caller's order, and composite_b probes first. That probe returns roughly 976k candidates against a 100-row batch, so the loop cannot stop and composite_a is probed as well. Swapping the two columns puts composite_a first, which returns 100 candidates and stops the loop after one probe.

Release builds, same machine, one sitting, swapping only the binary. Median of 5 runs after a discarded warmup. Cold means a fresh dataset handle per iteration, so the index cache is not primed.

case	main (concurrent probes)	this PR (sequential probes)
both probes needed, cold cache	36.6 ms	43.4 ms
both probes needed, warm cache	10.6 ms	11.6 ms
one probe suffices, cold cache	37.1 ms	3.5 ms
one probe suffices, warm cache	10.2 ms	2.4 ms

The worst case costs 19% cold and 9% warm. The overlap try_join! bought only covers the index probes, and even on a cold cache the take and the join dominate, so serializing two probe round trips moves a small part of the total.

One check that the construction measures what it claims: on main the two column orders come out the same (36.6 and 37.1 cold, 10.6 and 10.2 warm), which is what you would expect when every probe is always evaluated. The split only appears with this PR.

I have folded these numbers into the description as well.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

[LuciferYang]

cc @westonpace @wjones127 @Xuanwo FYI

[LuciferYang]

Agreed on the reading, and worth naming the condition under which the 19% shows up, because it is narrower than "two probes needed".

The regression needs the selectivity estimates to tie. The ordering is computed from the source side, so with 100 distinct source values in both key columns the two keys score the same, the stable sort keeps the caller's column order, and whichever key the caller happened to list first gets probed first. In the benchmark that is composite_b, which returns about 976k candidates against a 100-row batch, so the loop cannot stop and both probes run sequentially. When the estimates actually differ, the selective key goes first and the second probe never happens, which is where the 10x comes from.

So the cost is paid only when we had no information to order by and guessed wrong. That is why I left it: on main both column orders cost the same because every probe always runs, and this PR turns that into "much faster when we can tell the keys apart, slightly slower when we cannot."

If you would rather not pay it at all, the targeted mitigation is to keep the sequential loop only when the estimates distinguish the keys, and fall back to probing concurrently when they tie. That keeps the early-stop win, since it depends on one key being visibly more selective, and removes the sequential round trip in exactly the case that regressed. Happy to fold that into this PR or file it as a follow-up, whichever you prefer.

[lance-gatekeeper]

⚠️ Gate recommendation: approve with a non-blocking risk.

The latest clarification establishes that the measured 19% regression used a source-distinct-count tie. The implementation still runs probes sequentially whenever the first result exceeds the source batch size, so unequal estimates can also take the no-stop path; a tie-only concurrent fallback would remove the demonstrated regression, but not every possible one. Exactness and fragment coverage remain intact, and the full scan remains available for affected workloads, so this is still a non-blocking performance risk. Maintainers can accept it as-is or request the targeted fallback to eliminate the measured case.

[wjones127]

Nice work!

[LuciferYang]

Thank you @wjones127