lance-format / lance-format/lance

merge_insert is loading too many columns during indexed merge insert

Open
#3,480 2 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

enhancement priority: high
Dominant language
Rust
Stars
7.1k
Forks
852
Avg merge
3d 18h
Merged PRs (30d)
272

Description

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

Contributor guide

Open the contributing guide

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start in rust/lance/src/dataset/write/merge_insert.rs by comparing execute_uncommitted_impl, execute_uncommitted_v2, and create_plan, then inspect the query-plan test near lines 3001-3002. Implement the four TODO items and verify that up-to-date and partially indexed plans match the issue's expected shapes, with tests covering both cases.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
database
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.