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."""