Two‑Tier Manifest Architecture to Enable Lock‑free Parallel Column Writes

[hiltonhe]

Background & Problem

In large‑scale data production pipelines for LLM training, we use Ray to write many columns into a Lance dataset in parallel. Each Ray worker is exclusively responsible for appending data to a distinct column, so there is no data‑level write conflict – only metadata (manifest) conflict.

With the current single‑manifest design we face a dilemma:

• Frequent column‑level commits → Each worker commits its own column immediately after writing. Because Lance's optimistic concurrency control serialises all commits, this leads to high commit contention, many retries, rapid version inflation, and wasted metadata I/O.
• Single batched commit → We wait for all columns to finish and then issue one commit through a coordinator. This turns an otherwise fully parallel write phase into a synchronous blocking step. If any column write fails, the commit is delayed, and the entire batch must be retried.

Both paths limit the throughput and scalability that the underlying columnar storage format could otherwise deliver.

Proposed Solution: Two‑Tier Manifest

Split the global manifest into two layers:

1. GlobalManifest

• Holds the table schema (which is relatively static) and a column_pointers map.
• Each pointer references the latest ColumnManifest file for a given column, together with a column‑local version number.
• The global manifest remains small and is updated only when the schema changes or at the end of a coordinated write batch.

2. ColumnManifest (one per column)

• Each file contains the fragment list belonging to a single column.
• Writers for a column update their own ColumnManifest independently and without any global lock.
• Because each column is owned by exactly one worker during a write job, there is no conflict on the column manifest.

Expected Benefits

• Eliminates version inflation – Exactly one global version is created per full multi‑column write, regardless of the number of columns written.
• No commit contention – Column manifests are written without any lock; only the single GlobalManifest commit uses OCC, and that happens from a single coordinator.
• Column‑level fault isolation – A failed column write can be retried individually without blocking or invalidating the work of other columns.
• Retains full parallel throughput – Data writing and column‑manifest creation happen entirely in parallel; the only synchronisation point is a lightweight manifest merge.
• Drop‑in for many use‑cases – This aligns with the “fragments already written” model, just organising metadata differently.

[westonpace]

Frequent column‑level commits → Each worker commits its own column immediately after writing. Because Lance's optimistic concurrency control serialises all commits, this leads to high commit contention, many retries, rapid version inflation, and wasted metadata I/O.

I'm not sure I understand this point. Are you saying that you do one commit per fragment?

What operation is Ray using to do the commit? Is it using Merge or DataReplacement? The Merge operation is a bit legacy and all these commits will conflict. However, the DataReplacement operation is newer and will only conflict if you touch the same fragment & column. That would give you fragment-level commits without conflicts. Would this work?

[hiltonhe]

Frequent column‑level commits → Each worker commits its own column immediately after writing. Because Lance's optimistic concurrency control serialises all commits, this leads to high commit contention, many retries, rapid version inflation, and wasted metadata I/O.

I'm not sure I understand this point. Are you saying that you do one commit per fragment?

What operation is Ray using to do the commit? Is it using Merge or DataReplacement? The Merge operation is a bit legacy and all these commits will conflict. However, the DataReplacement operation is newer and will only conflict if you touch the same fragment & column. That would give you fragment-level commits without conflicts. Would this work?

Thank you very much for the clarification – I wasn't aware of the difference between Merge and DataReplacement, and the fragment‑level conflict detection sounds very promising.

To give you more context on our side:


• Initially, we tried having each Ray worker write its portion of a column and then immediately commit. I suspect we were using Merge (or perhaps a similar operation that conflicts globally), because we hit severe commit contention and version inflation – almost every commit would fail and need to retry, which made the approach unusable.

• To work around this, we switched to a coordinator‑based design: each worker writes its column data and records a checkpoint, but does not commit. Once all columns have finished writing, the coordinator performs a single commit for the entire table. This solves the version inflation and conflict problems, but it introduces a synchronisation point – we lose some parallelism and the commit is delayed if even one column write takes longer.

Your mention of DataReplacement suggests we might be able to get the best of both worlds: each worker could independently commit its own column (on distinct fragments, touching no other columns) without conflicts and without the need for a final blocking commit. If I understand correctly, the global version would still only advance once per real commit, with no wasted retries.

[westonpace]

I wasn't aware of the difference between Merge and DataReplacement, and the fragment‑level conflict detection sounds very promising.

It's semi-new. We are missing good python bindings for this operation, all we have is support for Dataset.commit I think. However, I'm pretty sure Ray uses the Dataset.commit path. I don't think any of the python APIs use it by default, but we use it quite a bit at our company for this kind of thing.

There is an example here:

lance/python/python/tests/test_dataset.py

Line 4521 in 57d2aad

def test_data_replacement(tmp_path: Path):

If I understand correctly, the global version would still only advance once per real commit, with no wasted retries.

I'm not sure what you mean by "real commit". The version would increase each time you do a commit but yes, there shouldn't be conflicts, as long as no two commits modified the same fragment. If you committed one fragment at a time.

You could also include more than one fragment per operation if you want and you'd get fewer commits and version increases.

[hiltonhe]

I wasn't aware of the difference between Merge and DataReplacement, and the fragment‑level conflict detection sounds very promising.

It's semi-new. We are missing good python bindings for this operation, all we have is support for Dataset.commit I think. However, I'm pretty sure Ray uses the Dataset.commit path. I don't think any of the python APIs use it by default, but we use it quite a bit at our company for this kind of thing.

There is an example here:

lance/python/python/tests/test_dataset.py

Line 4521 in 57d2aad

def test_data_replacement(tmp_path: Path):

If I understand correctly, the global version would still only advance once per real commit, with no wasted retries.

I'm not sure what you mean by "real commit". The version would increase each time you do a commit but yes, there shouldn't be conflicts, as long as no two commits modified the same fragment. If you committed one fragment at a time.

You could also include more than one fragment per operation if you want and you'd get fewer commits and version increases.

Thank you very much for your reply. Perhaps the previous description was too cumbersome, leading to some gaps

If we encounter the following scenarios, may I ask if there are better ways to handle them:

Currently, there are ten users, and the lance table has exactly ten columns. Each user is responsible for updating one of these columns. Since there is no mutual dependency between the columns during processing, we aim to handle them concurrently. Therefore, during the data production process, after User A processes column a, we hope to immediately perform a commit, which can then be passed on to the next step in the pipeline for reading. However, at this point, User B's data may not have finished running yet, making it impossible to commit and refresh only one version for each column update immediately.