lance-format / lance-format/lance-trino

Future: worker-distributed create_index build (blocked on lance-format/lance#5359)

Open
#189 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
22
Forks
16
Avg merge
10h 34m
Merged PRs (30d)
20

Description

## Motivation

#187/#188 added `ALTER TABLE ... EXECUTE create_index(...)`, currently parallelized across threads on the Trino **coordinator** only (each thread opens its own `Dataset` handle and builds one batch of fragments; see #188 for why sharing one handle across threads is actually slower). The natural follow-up: could this be distributed across separate Trino **worker** nodes instead, for real multi-machine scale?

This issue documents that investigation. The Trino-side mechanics are solved. The actual blocker is upstream in Lance's Java bindings, tracked at [lance-format/lance#5359](https://github.com/lance-format/lance/issues/5359) (open, unresolved) — this issue exists so the design is ready to build the moment that's unblocked, rather than being rediscovered from scratch later.

## The Trino-side mechanics (solved)

**Primary: reuse `ALTER TABLE ... EXECUTE`, no new SPI surface.**

`ConnectorPageSinkProvider` already has a default method:
```java
ConnectorPageSink createPageSink(ConnectorTransactionHandle, ConnectorSession, ConnectorTableExecuteHandle, ConnectorPageSinkId)
```
already reachable from `LancePageSinkProvider` today. Trino's `TableProcedureExecutionMode.distributedWithFilteringAndRepartitioning()` mode requires the table-execute's source scan to cover every column of the table — confirmed via `io.trino.sql.planner.LogicalPlanner#createTableExecutePlan` (trino tag 476):
```java
List columnNames = tableMetadata.columns().stream()... // ALL non-hidden columns
List symbols = visibleFields(tableScanPlan); // ALL scanned columns
verify(columnNames.size() == symbols.size(), ...);
```
This is a hard invariant on the **plan shape** (column count), not on how expensive reading those columns actually is. Since the connector controls both the page source and page sink for its own table-execute context, the scan can return placeholder rows — the indexed `VARCHAR` column carries an encoded payload instead of real text — while the real `Dataset.open(tablePath, ...).createIndex(fragmentIds=batch)` call happens as a side effect on the worker assigned that split. (Checked `RemoveEmptyTableExecute`, the one optimizer rule touching table-execute plan shape — it only fires when the source is statically known empty via predicate elimination, not relevant here; there could be others I haven't found.)

Net effect: the exact same SQL users already have — `ALTER TABLE lance.db.docs EXECUTE create_index(column => 'body', index_type => 'fts')` — genuinely distributed, no new syntax.

**Fallback: `ConnectorTableFunction` (leaf function) + connector-registered aggregate.**

Directly precedented by Trino's own `SequenceFunction` (`core/trino-main/.../operator/table/SequenceFunction.java`): a leaf table function (scalar arguments only, output schema independent of any base table via `DescribedTable`) that defines multiple splits, each processed by a `TableFunctionSplitProcessor` invoked via `TableFunctionProcessorProvider#getSplitProcessor()`, executed by `LeafTableFunctionOperator` on the worker assigned that split. A coordinator-side finalize step (merge + commit) can be a connector-registered aggregate/scalar function via `Connector#getFunctionProvider()` returning a `FunctionBundle` — precedented by `trino-datasketches`' `SketchFunctionsConnector`. Worse SQL ergonomics (`SELECT finalize(array_agg(...)) FROM TABLE(...)` instead of `ALTER TABLE ... EXECUTE`) and two new SPI extension points instead of zero, but proven if the placeholder-scan trick above hits something unforeseen.

Reusable entry points already in this repo for either approach: `LanceRuntime#openDatasetDirect(userIdentity, tablePath, version, storageOptions)` (independent `Dataset`, no shared coordinator state — exactly what a worker-side processor needs), `LanceRuntime#getFragments(...)` (fragment discovery for split partitioning), and `LanceTableExecuteHandle`'s record + `@JsonTypeInfo` pattern (template for a new split-handle class).

## The actual blocker: `org.lance.index.Index` can't leave the JVM process that created it

Both approaches above need a worker's `createIndex(fragmentIds=batch)` result to reach the coordinator for the final `mergeExistingIndexSegments` + `commitExistingIndexSegments`. That's not possible today:

- `Index` does not implement `Serializable` — unlike `FragmentMetadata`, which does, and which is exactly why the *existing* insert/write path can already ship fragment metadata between workers and the coordinator (`LanceMetadata.serializeFragment`/`deserializeFragment`). No equivalent exists for `Index`.
- `Index`'s constructor is private — no way to reconstruct one from raw field values either, so even a hand-rolled encoding couldn't produce a real `Index` to pass into `mergeExistingIndexSegments`/`commitExistingIndexSegments`.
- No alternate discovery API exists: `IndexCriteria`/`describeIndices()` only surface already-*committed* indices, nothing for "find the uncommitted segment with this UUID."

Confirmed via [lance-format/lance#5359](https://github.com/lance-format/lance/issues/5359) that this is a known, currently-unresolved upstream limitation (not something specific to lance-trino, and not new) — an open RFC from the Lance maintainers proposing a unified `DistributedIndexBuilder` specifically to "abstract the Index details away from the Python/Java bindings." Related: [#5164](https://github.com/lance-format/lance/issues/5164) documents a real, production-proven distributed BTree build via lance-spark + Spark's shuffle (130M rows/50 ranges → 3 min; 10B rows/1000 ranges → 15 min end-to-end, 46s merge) — strong evidence this matters at scale, though it's range-partitioned (Spark shuffle) rather than fragment-partitioned like ours, and its own merge step apparently has the same same-process-`Index` dependency per the #5359 quote above, so it isn't a ready-made pattern to copy.

I've left a comment on #5359 from the Trino-connector angle as a second concrete downstream consumer of whatever unification lands there.

## Why this is likely worth it once unblocked

- **Cloud storage is the common deployment**, not local disk — worker-node distribution is the relevant shape for real Lance deployments, not a niche one.
- **Network cost is very likely small by construction.** `Index`'s own fields (`uuid`, `fields`, `name`, `datasetVersion`, `fragments`, `indexDetails: byte[]`, `indexVersion`, `createdAt`, `baseId`, `indexType`) and its Javadoc ("corresponds to the Rust `Index` struct in `lance/rust/lance-table/src/format/index.rs`") indicate a manifest-level descriptor, the same role Iceberg's `ManifestFile`/`DataFile` entries play. `createIndex(fragmentIds=...)` already doesn't commit while still returning a real `Index`, meaning the actual segment content is almost certainly already durably written to shared/cloud storage by the worker, and only this small descriptor needs to reach the coordinator. Should be confirmed empirically (actual `indexDetails` byte size) once buildable, but it's a well-supported expectation.
- **Concrete scale motivation:** this session's benchmark (1M rows × 20 words/doc ≈ 100MB text) took ≈0.7s single-threaded. Naive linear extrapolation to a 1TB corpus (10,000x) suggests ≈2 hours on one coordinator core; the coordinator-local fix in #188 brings that to perhaps 60-90 minutes on one machine. #5164's real numbers above show the same shape at real scale: hours become minutes once genuinely distributed. A one-time or periodic full-text index build over a real large corpus (large document/log/email archive) stretching into hours on one machine is a legitimate trigger to revisit this, not a hypothetical one.

## Status

Not being built now. Trino-side design is ready; blocked on lance-format/lance#5359 (or whatever supersedes it) landing a cross-process-safe way to hand back index-segment metadata from a worker to a coordinator.

Contributor guide

Open the contributing guide

Research direction

Start with the upstream lance-format/lance#5359 blocker, then inspect LanceRuntime#openDatasetDirect, LanceRuntime#getFragments, and LanceTableExecuteHandle. The work is done when a worker can return index-segment metadata across processes so the existing distributed ALTER TABLE EXECUTE path can merge and commit the index on the coordinator.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
databases, distributed-systems
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
20/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.