fix: deduplicate row addresses in take to prevent panic

[wjones127]

Previously, duplicate row addresses (e.g. from FTS on List<Utf8> where
multiple list elements in the same row match) were passed to the v2 encoding
layer, which requires strictly increasing indices. This caused
indices_to_ranges to produce overlapping ranges, panicking with "attempt to
subtract with overflow".

Dedup is handled at two sites:

• FragmentReader::take_as_batch: This is the common entry point for all
  callers that need a single batch (FileFragment::take_rows,
  FragmentSession::take_rows, TakeStream::map_batch). Since it already
  collects into one batch, the expand-after-dedup step is trivial. Handling it
  here rather than in the v2 adapter's take_all_tasks avoids complications
  with the streaming ReadBatchTaskStream return type. Handling it lower in
  schedule_take wouldn't work because the encoding layer has no notion of
  duplicates by design.

• TakeStream::map_batch: Dedup here serves a different purpose — it
  reduces I/O by avoiding redundant reads of the same row across fragments.
  This also optimizes the common path (sorted, unique addresses) to skip all
  sorting/dedup/permutation overhead entirely.

Fixes #5260

Test plan

• test_take_with_duplicate_row_addrs — sorted duplicate row addresses
• test_take_with_unsorted_duplicate_row_addrs — unsorted duplicates
• test_fragment_take_indices / test_fragment_take_rows — fragment-level
  duplicate indices (previously failing on LanceFileVersion::Stable)
• All 57 take-related tests pass

🤖 Generated with Claude Code

[github-actions]

Code Review

This PR correctly fixes issue #5260 by deduplicating row addresses before passing them to the encoding layer which requires strictly increasing indices.

P1 Issue: Use proper error propagation instead of .unwrap()

At line 290 in the updated code, .unwrap() on take_record_batch could panic if it fails. While the existing code on line 272 also uses .unwrap() for the permutation restore, adding proper error propagation with ? would be more defensive.

Consider changing to:

new_data = arrow_select::take::take_record_batch(&new_data, &expand_indices)
    .map_err(|e| DataFusionError::Execution(e.to_string()))?;

The same applies to the existing .unwrap() on line 272, though that is outside the scope of this PR.

---

Overall the fix is correct and well-tested. The dedup approach is sound: using the sorted order to efficiently track unique addresses and then expanding results back with duplicates is the right pattern.

[codecov]

Codecov Report

❌ Patch coverage is 94.53125% with 7 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance/src/io/exec/take.rs	95.45%	5 Missing ⚠️
rust/lance/src/dataset/fragment.rs	88.23%	0 Missing and 2 partials ⚠️

📢 Thoughts on this report? Let us know!