merge_insert is loading too many columns during indexed merge insert

[westonpace]

We are loading all columns from the target table. We should only need to load the join key columns and/or any columns needed for a filter.

Background

We are migrating from an implementation of merge insert that manually manipulates streams to one that constructs a DataFusion plan and runs optimizers on that plan for optimize it.

The old code path is in execute_uncommitted_impl:

https://github.com/lancedb/lance/blob/94dcaf5e735480f9f61216b4e4f90358dbeac2ba/rust/lance/src/dataset/write/merge_insert.rs#L1337-L1340

While the new codepath is under execute_uncommitted_v2:

https://github.com/lancedb/lance/blob/94dcaf5e735480f9f61216b4e4f90358dbeac2ba/rust/lance/src/dataset/write/merge_insert.rs#L1352-L1371

Task

Goal: use the new code path when has_scalar_index=True.

If the index is up-to-date (no unindexed fragments), the plan should look like:

MergeInsert: on=[key], ...
  ProjectionExec: <assign action>
    AddRowAddrExec: ...
      ScalarIndexJoinExec: on=[key] index=[key_idx]
        StreamingTableExec: ...

Basically, the input should "join" against the index to find which do and don't have _rowids (exist in the dataset). Then we can use AddRowAddrExec to get the addresses.

If the index is not up-to-date, then the plan should be a union of the existing plan plus the indexed one, with an additional filter to ensure we don't duplicate new rows:

MergeInsert: on=[key], ...
  ProjectionExec: <assign action>
    UnionExec:
      AddRowAddrExec: ...
        FilterExec: _rowid is not null
          ScalarIndexJoinExec: on=[key] index=[key_idx]
            ReplayExec: ...
      HashJoinExec: on=[key]
        LanceScan: projection=[key], row_addr=true, ...
        ReplayExec: ...

Note: we have an existing test for query plans here you can imitate: https://github.com/lancedb/lance/blob/94dcaf5e735480f9f61216b4e4f90358dbeac2ba/rust/lance/src/dataset/write/merge_insert.rs#L3001-L3002

TODO

• Implement ScalarIndexJoinExec
• Implement ScalarIndexJoinNode (logical plan) with UserDefinedLocalNodeCore trait
• Implement ExtensionPlanner for ScalarIndexJoin that decides between up-to-date and not up-to-cade plans
• Update create_plan to handle creation of plan with ScalarIndexJoin

[pratik0316]

Hi @westonpace , @wjones127 have created a one-pager comparing the approaches https://docs.google.com/document/d/1NHxR2fqRF08KdRFv6Qr1lEaHEwTtfwsO8PuLXMEjAFk/edit?usp=sharing

of this issue and #6443

Please provide a review on the same

[wjones127]

FYI I've created a PR adding benchmarks, which should aid in this work: #8052

Will make it easy to see if we are improving things or creating regressions as we propose new designs.