lance-format / lance-format/lance
support distributed splits planning
Nobody has claimed this yet.
- Dominant language
- Rust
- Stars
- 7.1k
- Forks
- 852
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 272
Description
We had some discussions here and there about this API, and with development in quite a few distributed engines going on, it's probably the right time we start to actually work on it.
previous discussion: https://github.com/lance-format/lance/issues/4163#issuecomment-3050483230
For the read path a distributed engine with coordinator and worker architecture has the planning phase and distributed execution phase where each worker works on a unit of work (what we call Split) and then collect results back.
Within each node, we just pushdown the execution to Lance core SDK scanner in python/java, which does its independent predicate pushdown, column projection, aggregate pushdown, etc. using rust core scanner.
Today, the distribution logic is pretty simple: fragment is the split. In Spark, Trino, etc. it all just distribute by fragment directly. For Ray, we do something a bit more clever that we combine multiple fragments in a single unit of work, but it's not really smart either, it does not consider things like fragment size at all (https://github.com/lance-format/lance-ray/blob/main/lance_ray/utils.py#L140)
the plan_splits API is supposed to:
- get the requirements (predicate pushdown, column projection, aggregate pushdown, time travel, version travel, etc.)
- return an iterable of splits
where each split should contain the following information:
- the list of fragments to scan
- for each fragment, the start row position + the number of rows
- the residual predicate
- aggregation to apply
And when planning splits, we should take into consideration of the available indexes (btree, zone map, etc.), the index's fragment coverage, the fragment size, number of fragments to provide an ideal split distribution.
An example is a count + filter situation we are working on in Spark, consider a query SELECT count(*) FROM table WHERE id LIKE "abc%", and there is a btree index on id column. There are 10k fragments in the table, fragment 1 to 9000 are covered by index.
An ideal split plan in my mind would look like:
- split 1: fragment 0-3000 - indexed query of 3000 fragments per split
- split 2: fragment 3001-6000
- split 3: fragment 6001-9000
- split 4: fragment 9001-9100 - unindexed query of 100 fagments per split
- split 5: fragment 9101-9200
- ...
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the previous discussion linked in the issue and the existing grouping logic in lance_ray/utils.py around line 140. Define how the plan_splits API receives scan requirements and returns splits containing fragments, row ranges, residual predicates, and aggregations. Done means planning accounts for index coverage, fragment size, and balanced distribution across indexed and unindexed work.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, python, rust
- Domain
- backend-api-design, data-engineering, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 30/100