From 046987bfa13837e9295dc677e1fc4fd6967bd6d5 Mon Sep 17 00:00:00 2001 From: jasonz-lance Date: Thu, 8 Oct 2026 17:02:17 -0700 Subject: [PATCH] fix(python): reject LSM-enabled tables in 2-phase (block_size) reads A table with an LSM write spec may have acknowledged merge_insert writes that exist only in the MemWAL, not yet flushed into the base table/Lance fragments. _read_block's native scanner/take paths read the flushed base table directly at a pinned version, so they are blind to those unflushed rows -- no error, just silently stale values for the affected rows. 1-phase already refuses this (unflushed rows have no row id, so PermutationBuilder rejects it), but 2-phase's block reads are purely positional/offset-based and never touch row ids, so the same table sails through 2-phase construction and reads stale data on every affected row. Reject at block_size construction time instead, mirroring the existing non-default-branch guard immediately above it. Verified with gatekeeper's exact reproducer (8-row table, LsmWriteSpec.bucket + merge_insert, read back through StreamingDataset) adapted into a pytest regression. Co-Authored-By: Claude Sonnet 5 --- python/python/lancedb/streaming.py | 20 ++++++++++++- .../python/tests/test_elastic_dataloader.py | 30 +++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/python/python/lancedb/streaming.py b/python/python/lancedb/streaming.py index 22135ac30..530a70d9c 100644 --- a/python/python/lancedb/streaming.py +++ b/python/python/lancedb/streaming.py @@ -451,7 +451,7 @@ class StreamingDataset(IterableDataset): default, the number of workers is determined by ``os.cpu_count()``. The main thread round-robins over the cooked queues, yielding one row per - split per cycle. + split per cycle. Parameters ---------- @@ -895,6 +895,24 @@ class StreamingDataset(IterableDataset): f"a non-default branch ({self._pinned_branch!r}); check " "out the default branch before constructing StreamingDataset" ) + # An LSM write spec means some acknowledged writes only exist in + # the MemWAL, not yet flushed into the base table that the native + # scanner/take paths below read; a version pin can't see them, so + # reads here would silently omit them with no error. 1-phase + # already refuses this for the same reason (unflushed rows have + # no row id -- see PermutationBuilder); block_size mode has no + # row ids in its read path either, but reads positionally, so it + # fails differently (stale values, not a missing-row-id error) + # and needs its own explicit check. + if table.get_lsm_write_spec() is not None: + raise ValueError( + "block_size does not yet support a table with an LSM " + "write spec set; unflushed MemWAL rows are not visible " + "to a version-pinned read, so block reads could " + "silently miss acknowledged writes. Call " + "table.checkpoint_lsm() and table.unset_lsm_write_spec() " + "first, or construct StreamingDataset without block_size" + ) self._pinned_version = table.version block_live_offsets: Optional[list[np.ndarray]] = None diff --git a/python/python/tests/test_elastic_dataloader.py b/python/python/tests/test_elastic_dataloader.py index e072c7e83..da1eeb863 100644 --- a/python/python/tests/test_elastic_dataloader.py +++ b/python/python/tests/test_elastic_dataloader.py @@ -3418,6 +3418,36 @@ def test_two_phase_pinned_version_survives_concurrent_write(tmp_path): ) +def test_two_phase_rejects_lsm_write_spec_table(tmp_path): + """An LSM write spec means some acknowledged writes live only in the + MemWAL, invisible to a version-pinned read; block_size must refuse the + table at construction rather than silently reading stale base-table + values (mirroring the row-id rejection 1-phase already applies).""" + from lancedb import LsmWriteSpec + + db = lancedb.connect(tmp_path) + schema = pa.schema( + [ + pa.field("id", pa.int64(), nullable=False), + pa.field("value", pa.int64(), nullable=False), + ] + ) + table = db.create_table( + "t", + pa.Table.from_pydict({"id": list(range(8)), "value": [0] * 8}, schema=schema), + ) + table.set_unenforced_primary_key("id") + table.set_lsm_write_spec(LsmWriteSpec.bucket("id", 1)) + table.merge_insert( + "id" + ).when_matched_update_all().when_not_matched_insert_all().execute( + pa.Table.from_pydict({"id": list(range(8)), "value": [99] * 8}, schema=schema) + ) + + with pytest.raises(ValueError, match="LSM write spec"): + StreamingDataset(table, num_splits=1, block_size=4, shuffle=False) + + def test_two_phase_reshuffle_blocks_per_epoch_toggle(lance_table): """reshuffle_blocks_per_epoch=False keeps each split's blocks fixed across epochs; the default (True) reassigns them every epoch."""