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 <noreply@anthropic.com>

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

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