chore: update lance dependency to v9.0.0-beta.16

### **Summary**
Closes #3212

Extends the Python `lit()` helper to natively support three additional
types (`date`, `datetime`, and `Decimal`) and implements reflexive
operators for the `Expr` class.

This implementation specifically addresses the blocking feedback
regarding precision loss, CI discovery, and query engine limitations:

* **Logic Refactoring**: Simplified `lit()` by combining `date` and
`datetime` normalization into ISO-8601 strings, ensuring stable SQL
parsing across different engine locales.
* **Precision Preservation**: `decimal.Decimal` objects are now passed
as high-precision strings to the Rust bridge, bypassing intermediate
float conversions and preserving full 128-bit decimal precision for
DataFusion.
* **Averted CI Failures**: Temporarily deferred `bytes` literal support
to a future PR to resolve a known DataFusion `expr_to_sql` limitation
that was crashing the `Doctest` runner.
* **Reflexive Operators**: Added support for "literal-first" arithmetic
and logical operations (e.g., `10 + col('a')` or `True &
col('active')`). Redundant reflexive comparisons (e.g., `__rlt__`) were
pruned as Python's data model handles them automatically.
* **Integration Verification**: Added dedicated integration tests in the
official test directory to ensure the query engine correctly handles the
new types and preserves bit-perfect fidelity.

### **Changes**  
####
[python/python/lancedb/expr.py](file:///c:/Users/Laksh/Documents/lancedb/python/python/lancedb/expr.py)
* Updated `lit()` to handle `date`, `datetime`, and `Decimal` natively.
* Implemented reflexive operators (`__radd__`, `__rand__`, `__rmul__`,
etc.) to support literals on the left-hand side.
* Removed the problematic `bytes` doctest example and `lit()` type
support to unblock CI.

####
[python/src/expr.rs](file:///c:/Users/Laksh/Documents/lancedb/python/src/expr.rs)
* Modified the Rust FFI bridge to extract `Decimal` objects as strings.
* Ensured the `expr_lit` handler is ready to receive normalized temporal
strings.
*   Consolidated imports and added missing operator documentation.

####
[python/python/lancedb/_lancedb.pyi](file:///c:/Users/Laksh/Documents/lancedb/python/python/lancedb/_lancedb.pyi)
* Updated type stubs for `expr_lit` to include `Any` (allowing for
`Decimal`).

### **Testing**  
Added several new advanced test cases in
[python/python/tests/test_expr.py](file:///c:/Users/Laksh/Documents/lancedb/python/python/tests/test_expr.py)
covering:
* **High-precision Decimal preservation**: Verified against 128-bit
boundaries with a "one point off" test case (`1.234567890123456789 <
1.234567890123456790`).
* **Reflexive operator positioning**: Verified successful query
construction with literals on the left.
* **Timezone-aware normalization**: Confirmed stable behavior for
`datetime` objects.
* **Integration Testing**: Confirmed Date32 and Decimal columns return
the correct Python types and values from the engine during `.to_arrow()`
calls.

---------

Co-authored-by: Will Jones <willjones127@gmail.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.31.0-beta.5"
+current_version = "0.31.0-beta.6"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

Cargo.lock

@@ -3423,8 +3423,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
 [[package]]
 name = "fsst"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "rand 0.9.4",
@@ -4726,8 +4726,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
 
 [[package]]
 name = "lance"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -4801,8 +4801,8 @@ dependencies = [
 
 [[package]]
 name = "lance-arrow"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4823,7 +4823,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-scalar"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4837,7 +4837,7 @@ dependencies = [
 [[package]]
 name = "lance-arrow-stats"
 version = "58.0.0"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-schema",
@@ -4846,8 +4846,8 @@ dependencies = [
 
 [[package]]
 name = "lance-bitpacking"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrayref",
  "crunchy",
@@ -4857,8 +4857,8 @@ dependencies = [
 
 [[package]]
 name = "lance-core"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -4896,8 +4896,8 @@ dependencies = [
 
 [[package]]
 name = "lance-datafusion"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -4927,8 +4927,8 @@ dependencies = [
 
 [[package]]
 name = "lance-datagen"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -4945,8 +4945,8 @@ dependencies = [
 
 [[package]]
 name = "lance-derive"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -4955,8 +4955,8 @@ dependencies = [
 
 [[package]]
 name = "lance-encoding"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -4991,8 +4991,8 @@ dependencies = [
 
 [[package]]
 name = "lance-file"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-arith",
  "arrow-array",
@@ -5022,8 +5022,8 @@ dependencies = [
 
 [[package]]
 name = "lance-index"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arc-swap",
  "arrow",
@@ -5088,8 +5088,8 @@ dependencies = [
 
 [[package]]
 name = "lance-io"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow",
  "arrow-arith",
@@ -5130,8 +5130,8 @@ dependencies = [
 
 [[package]]
 name = "lance-linalg"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5147,8 +5147,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow",
  "async-trait",
@@ -5160,8 +5160,8 @@ dependencies = [
 
 [[package]]
 name = "lance-namespace-impls"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow",
  "arrow-ipc",
@@ -5215,8 +5215,8 @@ dependencies = [
 
 [[package]]
 name = "lance-select"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5231,8 +5231,8 @@ dependencies = [
 
 [[package]]
 name = "lance-table"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow",
  "arrow-array",
@@ -5271,22 +5271,22 @@ dependencies = [
 
 [[package]]
 name = "lance-testing"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "arrow-array",
  "arrow-schema",
  "criterion",
  "lance-arrow",
  "num-traits",
- "pprof",
+ "pprof 0.15.0",
  "rand 0.9.4",
 ]
 
 [[package]]
 name = "lance-tokenizer"
-version = "9.0.0-beta.10"
+version = "9.0.0-beta.16"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.16#78a814b6b448b433f5ebfdb9a8dbdec7b904c59f"
 dependencies = [
  "icu_segmenter",
  "jieba-rs",
@@ -5299,7 +5299,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb"
-version = "0.31.0-beta.5"
+version = "0.31.0-beta.6"
 dependencies = [
  "ahash",
  "anyhow",
@@ -5361,6 +5361,7 @@ dependencies = [
  "pin-project",
  "polars",
  "polars-arrow",
+ "pprof 0.14.1",
  "rand 0.9.4",
  "random_word",
  "regex",
@@ -5384,7 +5385,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-nodejs"
-version = "0.31.0-beta.5"
+version = "0.31.0-beta.6"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5409,7 +5410,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-python"
-version = "0.34.0-beta.5"
+version = "0.34.0-beta.6"
 dependencies = [
  "arrow",
  "async-trait",
@@ -5952,9 +5953,9 @@ dependencies = [
 
 [[package]]
 name = "napi"
-version = "3.9.4"
+version = "3.10.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b41bda2ac390efb5e8d22025d925ccc3f3807d8c1bea6d19b36127247c4b8f83"
+checksum = "0c71997d6f7ad4a756966e452426848ac27d3b37a295302d63afbbcce0270f93"
 dependencies = [
  "bitflags 2.11.1",
  "chrono",
@@ -5977,9 +5978,9 @@ checksum = "c9c366d2c8c60b86fa632df75f745509b52f9128f91a6bad4c796e44abb505e1"
 
 [[package]]
 name = "napi-derive"
-version = "3.5.7"
+version = "3.5.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "61d66f70256ad5aef58659966064471d0ad90e2897bc36a5a5e0389c85aabc1e"
+checksum = "d4ba572deef53e2c386759a8c2014175a62679d74ff83adc205c8bc0e0285727"
 dependencies = [
  "convert_case",
  "ctor 1.0.5",
@@ -5991,9 +5992,9 @@ dependencies = [
 
 [[package]]
 name = "napi-derive-backend"
-version = "5.0.5"
+version = "5.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "81b4b08f15eed7a2a20c3f4c6314013fc3ac890a3afa9892b594485299ebdb2d"
+checksum = "ddd961eb2aa8965e3f29722d754f3a86907eb1984e2fbcbe3fe87b9a02d6bfba"
 dependencies = [
  "convert_case",
  "proc-macro2",
@@ -7326,6 +7327,28 @@ version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
 
+[[package]]
+name = "pprof"
+version = "0.14.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afad4d4df7b31280028245f152d5a575083e2abb822d05736f5e47653e77689f"
+dependencies = [
+ "aligned-vec",
+ "backtrace",
+ "cfg-if 1.0.4",
+ "findshlibs",
+ "inferno",
+ "libc",
+ "log",
+ "nix",
+ "once_cell",
+ "smallvec",
+ "spin 0.10.0",
+ "symbolic-demangle",
+ "tempfile",
+ "thiserror 1.0.69",
+]
+
 [[package]]
 name = "pprof"
 version = "0.15.0"

Cargo.toml

@@ -13,20 +13,20 @@ categories = ["database-implementations"]
 rust-version = "1.91.0"
 
 [workspace.dependencies]
-lance = { "version" = "=9.0.0-beta.10", default-features = false, "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance = { "version" = "=9.0.0-beta.16", default-features = false, "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-core = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-core = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-datagen = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-datagen = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-file = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-file = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-io = { "version" = "=9.0.0-beta.10", default-features = false, "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-io = { "version" = "=9.0.0-beta.16", default-features = false, "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-index = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-index = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-linalg = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-linalg = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-namespace-impls = { "version" = "=9.0.0-beta.10", default-features = false, "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-namespace-impls = { "version" = "=9.0.0-beta.16", default-features = false, "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-table = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-table = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-testing = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-testing = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-datafusion = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-datafusion = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-encoding = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-encoding = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
-lance-arrow = { "version" = "=9.0.0-beta.10", "tag" = "v9.0.0-beta.10", "git" = "https://github.com/lance-format/lance.git" }
+lance-arrow = { "version" = "=9.0.0-beta.16", "tag" = "v9.0.0-beta.16", "git" = "https://github.com/lance-format/lance.git" }
 ahash = "0.8"
 # Note that this one does not include pyarrow
 arrow = { version = "58.0.0", optional = false }

deny.toml

@@ -119,6 +119,18 @@ ignore = [
     # https://rustsec.org/advisories/RUSTSEC-2026-0177
     { id = "RUSTSEC-2026-0176", reason = "pyo3 in Python bindings; awaiting patched pyo3 release" },
     { id = "RUSTSEC-2026-0177", reason = "pyo3 in Python bindings; awaiting patched pyo3 release" },
+
+    # quick-xml < 0.41.0: quadratic runtime on duplicate attribute names (DoS).
+    # quick-xml < 0.41.0: unbounded namespace-declaration allocation in NsReader (DoS).
+    # Pulled in transitively by inferno (dev-only flame-graph dep), lance-namespace-impls
+    # (git dep from lance), and opendal/reqsign (cloud storage XML parsing). The XML
+    # parsed by opendal/reqsign comes from trusted cloud-storage endpoints (S3, GCS,
+    # Azure), not attacker-controlled input. Clearing requires upstream crates to migrate
+    # to quick-xml >= 0.41.0.
+    # https://rustsec.org/advisories/RUSTSEC-2026-0194
+    # https://rustsec.org/advisories/RUSTSEC-2026-0195
+    { id = "RUSTSEC-2026-0194", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
+    { id = "RUSTSEC-2026-0195", reason = "transitive via inferno/lance/opendal; XML from trusted cloud endpoints, not attacker-controlled" },
 ]
 
 # ---------------------------------------------------------------------------

docs/src/java/java.md

@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-core</artifactId>
-    <version>0.31.0-beta.5</version>
+    <version>0.31.0-beta.6</version>
 </dependency>
 ```
 

docs/src/js/interfaces/SplitRandomOptions.md

@@ -8,6 +8,14 @@
 
 ## Properties
 
+### clumpSize?
+
+```ts
+optional clumpSize: number;
+```
+
+***
+
 ### counts?
 
 ```ts

java/lancedb-core/pom.xml

@@ -8,7 +8,7 @@
     <parent>
       <groupId>com.lancedb</groupId>
       <artifactId>lancedb-parent</artifactId>
-      <version>0.31.0-beta.5</version>
+      <version>0.31.0-beta.6</version>
       <relativePath>../pom.xml</relativePath>
     </parent>
 

java/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-parent</artifactId>
-    <version>0.31.0-beta.5</version>
+    <version>0.31.0-beta.6</version>
     <packaging>pom</packaging>
     <name>${project.artifactId}</name>
     <description>LanceDB Java SDK Parent POM</description>
@@ -28,7 +28,7 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <arrow.version>15.0.0</arrow.version>
-        <lance-core.version>9.0.0-beta.10</lance-core.version>
+        <lance-core.version>9.0.0-beta.16</lance-core.version>
         <spotless.skip>false</spotless.skip>
         <spotless.version>2.30.0</spotless.version>
         <spotless.java.googlejavaformat.version>1.7</spotless.java.googlejavaformat.version>

nodejs/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.31.0-beta.5"
+version = "0.31.0-beta.6"
 publish = false
 license.workspace = true
 description.workspace = true

nodejs/npm/darwin-arm64/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.31.0-beta.5",
+	"version": "0.31.0-beta.6",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",

nodejs/npm/linux-arm64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.31.0-beta.5",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",

nodejs/npm/linux-arm64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.31.0-beta.5",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",

nodejs/npm/linux-x64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.31.0-beta.5",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",

nodejs/npm/linux-x64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.31.0-beta.5",
+	"version": "0.31.0-beta.6",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",

nodejs/npm/win32-arm64-msvc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.31.0-beta.5",
+  "version": "0.31.0-beta.6",
   "os": [
     "win32"
   ],

nodejs/npm/win32-x64-msvc/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.31.0-beta.5",
+	"version": "0.31.0-beta.6",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",

nodejs/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "@lancedb/lancedb",
-  "version": "0.31.0-beta.5",
+  "version": "0.31.0-beta.6",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@lancedb/lancedb",
-      "version": "0.31.0-beta.5",
+      "version": "0.31.0-beta.6",
       "cpu": [
         "x64",
         "arm64"

nodejs/package.json

@@ -11,7 +11,7 @@
     "ann"
   ],
   "private": false,
-  "version": "0.31.0-beta.5",
+  "version": "0.31.0-beta.6",
   "main": "dist/index.js",
   "exports": {
     ".": "./dist/index.js",

nodejs/src/permutation.rs

@@ -16,6 +16,7 @@ pub struct SplitRandomOptions {
     pub counts: Option<Vec<i64>>,
     pub fixed: Option<i64>,
     pub seed: Option<i64>,
+    pub clump_size: Option<i64>,
     pub split_names: Option<Vec<String>>,
 }
 
@@ -125,10 +126,15 @@ impl PermutationBuilder {
         };
 
         let seed = options.seed.map(|s| s as u64);
+        let clump_size = options.clump_size.map(|c| c as u64);
 
         self.modify(|builder| {
             builder.with_split_strategy(
-                SplitStrategy::Random { seed, sizes },
+                SplitStrategy::Random {
+                    seed,
+                    sizes,
+                    clump_size,
+                },
                 options.split_names.clone(),
             )
         })

python/benchmarks/bench_streaming_dataloader.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Benchmark for StreamingDataset throughput.
+
+Sweeps read_batch_size from 1 to 16384 to show how amortising the per-request
+overhead scales.  Each row at each chunk size is timed via the real
+StreamingDataset so the numbers reflect production code.
+
+Run with:
+    cd python
+    uv run --extra tests benchmarks/bench_streaming_dataloader.py
+
+Optional env vars:
+    BENCH_NUM_ROWS   — total rows in the table (default 49152 = 24 × 2048)
+    BENCH_NUM_SPLITS — number of splits (default 24)
+    BENCH_STEPS      — round-robin cycles to time per chunk size (default 100)
+    BENCH_ROW_BYTES  — approximate bytes per row padded with a binary column
+                       (default 4096, mimics a small embedding/image patch)
+"""
+
+import os
+import time
+import tempfile
+
+import pyarrow as pa
+import lancedb
+
+from lancedb.streaming import StreamingDataset
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+NUM_SPLITS = int(os.environ.get("BENCH_NUM_SPLITS", 24))
+# Default: 2048 rows per split so every chunk size up to 16Ki has ≥1 full
+# chunk (except 16Ki itself which gets a single full-split fetch — still valid).
+NUM_ROWS = int(os.environ.get("BENCH_NUM_ROWS", NUM_SPLITS * 2048))
+STEPS = int(os.environ.get("BENCH_STEPS", 100))
+ROW_BYTES = int(os.environ.get("BENCH_ROW_BYTES", 4096))
+
+assert NUM_ROWS % NUM_SPLITS == 0, "NUM_ROWS must be divisible by NUM_SPLITS"
+
+CHUNK_SIZES = [1, 4, 16, 64, 256, 1024, 4096, 16384]
+
+
+# ---------------------------------------------------------------------------
+# Table helpers
+# ---------------------------------------------------------------------------
+
+
+def make_table(db_path: str) -> lancedb.table.Table:
+    db = lancedb.connect(db_path)
+    payload = b"x" * ROW_BYTES
+    data = pa.table(
+        {
+            "id": pa.array(range(NUM_ROWS), type=pa.int32()),
+            "payload": pa.array([payload] * NUM_ROWS, type=pa.large_binary()),
+        }
+    )
+    return db.create_table("bench", data, mode="overwrite")
+
+
+# ---------------------------------------------------------------------------
+# Timing
+# ---------------------------------------------------------------------------
+
+
+def bench_chunk(table, chunk_size: int, steps: int) -> tuple[int, float]:
+    """Return (rows_drained, elapsed_seconds) for one timed run."""
+    total_rows = steps * NUM_SPLITS
+    ds = StreamingDataset(
+        table, num_splits=NUM_SPLITS, shuffle_seed=42, read_batch_size=chunk_size
+    )
+    count = 0
+    t0 = time.perf_counter()
+    for _ in ds:
+        count += 1
+        if count >= total_rows:
+            break
+    return count, time.perf_counter() - t0
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    rows_per_split = NUM_ROWS // NUM_SPLITS
+    print("Benchmark config:")
+    print(
+        f"  NUM_ROWS={NUM_ROWS}  NUM_SPLITS={NUM_SPLITS}  "
+        f"rows/split={rows_per_split}  STEPS={STEPS}  ROW_BYTES={ROW_BYTES}"
+    )
+    print(f"  ~{NUM_ROWS * ROW_BYTES / 1024 / 1024:.1f} MB total table size")
+    print()
+
+    with tempfile.TemporaryDirectory() as tmp:
+        print("Creating table...", flush=True)
+        table = make_table(tmp)
+
+        cols = (
+            f"{'chunk':>6}  {'rows':>6}  {'elapsed':>8}  {'rows/s':>10}  {'ms/step':>9}"
+        )
+        print(f"\n{cols}")
+        print("-" * 52)
+
+        for chunk in CHUNK_SIZES:
+            # Warm-up pass (one step's worth of rows)
+            warmup_ds = StreamingDataset(
+                table, num_splits=NUM_SPLITS, shuffle_seed=42, read_batch_size=chunk
+            )
+            warmup_count = 0
+            for _ in warmup_ds:
+                warmup_count += 1
+                if warmup_count >= NUM_SPLITS:
+                    break
+
+            drained, elapsed = bench_chunk(table, chunk, STEPS)
+            rows_per_sec = drained / elapsed if elapsed > 0 else float("inf")
+            ms_per_step = elapsed / STEPS * 1000
+
+            print(
+                f"{chunk:>6}  {drained:>6}  {elapsed:>7.3f}s  "
+                f"{rows_per_sec:>10.0f}  {ms_per_step:>8.1f}ms"
+            )
+
+    print()
+    print("Done.")
+
+
+if __name__ == "__main__":
+    main()

python/python/lancedb/_lancedb.pyi

@@ -1,4 +1,5 @@
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta
+from decimal import Decimal
 from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal
 
 import pyarrow as pa
@@ -53,7 +54,9 @@ class PyExpr:
     def to_sql(self) -> str: ...
 
 def expr_col(name: str) -> PyExpr: ...
-def expr_lit(value: Union[bool, int, float, str, bytes]) -> PyExpr: ...
+def expr_lit(
+    value: Union[bool, int, float, str, bytes, date, datetime, Decimal],
+) -> PyExpr: ...
 def expr_func(name: str, args: List[PyExpr]) -> PyExpr: ...
 
 class Session:

python/python/lancedb/expr.py

@@ -19,6 +19,8 @@ operators::
 
 from __future__ import annotations
 
+from datetime import date, datetime
+from decimal import Decimal
 from typing import Iterable, Union
 
 import pyarrow as pa
@@ -63,7 +65,7 @@ def _coerce(value: "ExprLike") -> "Expr":
 
 
 # Type alias used in annotations.
-ExprLike = Union["Expr", bool, int, float, str, bytes]
+ExprLike = Union["Expr", bool, int, float, str, bytes, date, datetime, Decimal]
 
 
 class Expr:
@@ -118,10 +120,18 @@ class Expr:
         """Logical AND (``expr_a & expr_b``)."""
         return Expr(self._inner.and_(_coerce(other)._inner))
 
+    def __rand__(self, other: ExprLike) -> "Expr":
+        """Right-hand logical AND (``True & expr``)."""
+        return Expr(_coerce(other)._inner.and_(self._inner))
+
     def __or__(self, other: "Expr") -> "Expr":
         """Logical OR (``expr_a | expr_b``)."""
         return Expr(self._inner.or_(_coerce(other)._inner))
 
+    def __ror__(self, other: ExprLike) -> "Expr":
+        """Right-hand logical OR (``False | expr``)."""
+        return Expr(_coerce(other)._inner.or_(self._inner))
+
     def __invert__(self) -> "Expr":
         """Logical NOT (``~expr``)."""
         return Expr(self._inner.not_())
@@ -266,13 +276,14 @@ def col(name: str) -> Expr:
     return Expr(expr_col(name))
 
 
-def lit(value: Union[bool, int, float, str, bytes]) -> Expr:
+def lit(value: Union[bool, int, float, str, bytes, date, datetime, Decimal]) -> Expr:
     """Create a literal (constant) value expression.
 
     Parameters
     ----------
     value:
-        A Python ``bool``, ``int``, ``float``, ``str``, or ``bytes``.
+        A Python ``bool``, ``int``, ``float``, ``str``, ``bytes``, ``date``,
+        ``datetime``, or ``Decimal``.
 
     Examples
     --------
@@ -280,6 +291,9 @@ def lit(value: Union[bool, int, float, str, bytes]) -> Expr:
     >>> col("price") * lit(1.1)
     Expr((price * 1.1))
     """
+    if not isinstance(value, (bool, int, float, str, bytes, date, datetime, Decimal)):
+        raise TypeError(f"Unsupported literal type: {type(value).__name__}")
+
     return Expr(expr_lit(value))
 
 

python/python/lancedb/permutation.py

@@ -65,6 +65,7 @@ class PermutationBuilder:
         counts: Optional[list[int]] = None,
         fixed: Optional[int] = None,
         seed: Optional[int] = None,
+        clump_size: Optional[int] = None,
         split_names: Optional[list[str]] = None,
     ) -> "PermutationBuilder":
         """
@@ -87,6 +88,9 @@ class PermutationBuilder:
         Rows will be randomly assigned to splits.  The optional seed can be provided to
         make the assignment deterministic.
 
+        If clump_size is provided, rows are shuffled as contiguous groups of that size,
+        preserving I/O locality while still randomising the split assignment.
+
         The optional split_names can be provided to name the splits.  If not provided,
         the splits can only be referenced by their index.
         """
@@ -95,6 +99,7 @@ class PermutationBuilder:
             counts=counts,
             fixed=fixed,
             seed=seed,
+            clump_size=clump_size,
             split_names=split_names,
         )
         return self

python/python/lancedb/streaming.py

@@ -0,0 +1,607 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+"""Elastic streaming dataloader for PyTorch.
+
+Provides StreamingDataset, a PyTorch IterableDataset that guarantees:
+
+- **Elastic determinism**: for a fixed (num_splits, shuffle_seed, epoch) the set
+  of samples that forms each global training step is identical regardless of
+  world_size or num_workers.
+- **Resumability**: state_dict / load_state_dict capture per-split consumption
+  counts so training can resume from an exact mid-epoch position even when the
+  distributed topology changes between runs.
+"""
+
+import ctypes
+import logging
+import os
+import random
+import threading
+import time
+from collections import deque
+from concurrent.futures import ThreadPoolExecutor
+from multiprocessing import RawArray
+from typing import Any, Callable, Iterator, Optional
+
+from torch.utils.data import IterableDataset, get_worker_info
+
+from .permutation import (
+    Permutation,
+    Transforms,
+    permutation_builder,
+    _table_from_pickle_state,
+    _table_to_pickle_state,
+)
+
+logger = logging.getLogger(__name__)
+
+# Multiplier used to combine shuffle_seed and epoch into a single permutation
+# seed.  Chosen to be a large prime so different (seed, epoch) pairs produce
+# distinct seeds for any practically encountered epoch count.
+_EPOCH_PRIME = 100003
+
+DEFAULT_READ_BATCH_SIZE = 64
+DEFAULT_PREFETCH_BATCHES = 4
+
+
+class StreamingDataset(IterableDataset):
+    """An elastic, resumable PyTorch IterableDataset backed by a LanceDB table.
+
+    The table is partitioned into ``num_splits`` fixed splits using a
+    deterministic random shuffle controlled by ``shuffle_seed`` and ``epoch``.
+    Each rank is assigned a contiguous block of splits, and within a rank each
+    DataLoader worker is assigned a contiguous sub-block.  Samples are yielded
+    by round-robining over the assigned splits, one sample per split per cycle.
+
+    Internally ``__iter__`` runs a two-stage pipeline:
+
+    - **Stage 1 (I/O)**: one thread pool with ``num_splits * prefetch_batches``
+      workers fetches raw ``RecordBatch`` objects from LanceDB in parallel
+      across all splits and places them in a per-split raw-batch queue.
+    - **Stage 2 (transform)**: a second thread pool with ``os.cpu_count()``
+      workers picks up raw batches, applies the transform, and places the
+      results in a per-split cooked-row queue.
+
+    The main thread round-robins over the cooked queues, yielding one row per
+    split per cycle.
+
+    Parameters
+    ----------
+    table:
+        LanceDB table to stream from.
+    num_splits:
+        Number of fixed splits to partition the table into.  Must be divisible
+        by ``world_size``.  When used with DataLoader workers it must also be
+        divisible by ``world_size * num_workers``.  Defaults to ``world_size``.
+        If the row count (after any ``filter``) is not evenly divisible by
+        ``num_splits``, the surplus rows — at most ``num_splits - 1`` per epoch
+        — are silently dropped to keep all splits the same length.
+    shuffle:
+        Whether to randomly assign rows to splits.  When ``True`` (the
+        default) rows are shuffled using ``shuffle_seed`` and ``epoch``.
+        When ``False`` rows are divided into splits sequentially in storage
+        order, which can be useful for deterministic debugging or evaluation.
+    shuffle_seed:
+        Base seed for the random permutation.  Combined with ``epoch`` so
+        each epoch produces a different ordering.  Pass ``None`` to generate
+        a random seed at construction time.
+    epoch:
+        Current training epoch.  Combined with ``shuffle_seed`` so that each
+        epoch produces a different sample ordering.
+    rank:
+        This process's rank in the distributed training group.
+    world_size:
+        Total number of processes in the distributed training group.
+    read_batch_size:
+        Number of rows fetched from each split in a single ``take_offsets``
+        call.  Larger values amortise per-request overhead (critical on object
+        storage) at the cost of higher memory usage per split buffer.  Defaults
+        to ``DEFAULT_READ_BATCH_SIZE`` (64).
+    prefetch_batches:
+        Number of I/O batches to keep in flight per split.  Higher values
+        overlap storage latency with transform and training compute at the cost
+        of more memory and threads.  Defaults to ``DEFAULT_PREFETCH_BATCHES``
+        (4).
+    columns:
+        Optional list of column names to read.  When set, only those columns
+        are fetched from storage; all others are omitted.  ``None`` (the
+        default) reads every column.
+    shuffle_clump_size:
+        When set, rows are shuffled in contiguous groups of this size rather
+        than individually.  Larger clumps improve I/O locality (important on
+        object storage) at the cost of reduced randomness.  ``None`` (the
+        default) shuffles rows individually.
+    filter:
+        Optional SQL filter expression (e.g. ``"label = 'dog'"``).  Only rows
+        that satisfy the predicate are included in the permutation.  The filter
+        is applied during permutation construction so split sizes reflect the
+        filtered row count.
+    transform:
+        Optional callable applied to each ``pyarrow.RecordBatch`` before rows
+        are yielded.  Receives one batch at a time and must return an iterable
+        whose length equals the number of rows in the batch.  When ``None``
+        (the default) rows are returned as plain Python dicts.
+    worker_info_override:
+        If set, used in place of ``torch.utils.data.get_worker_info()`` to
+        determine the DataLoader worker assignment.  Intended for unit tests
+        that need to simulate multiple workers without spawning real processes.
+        If both this and the real worker info are non-None a warning is logged
+        and the override takes precedence.
+    """
+
+    def __init__(
+        self,
+        table,
+        *,
+        num_splits: Optional[int] = None,
+        shuffle: bool = True,
+        shuffle_seed: Optional[int] = 0,
+        epoch: int = 0,
+        rank: int = 0,
+        world_size: int = 1,
+        read_batch_size: int = DEFAULT_READ_BATCH_SIZE,
+        prefetch_batches: int = DEFAULT_PREFETCH_BATCHES,
+        columns: Optional[list[str]] = None,
+        shuffle_clump_size: Optional[int] = None,
+        filter: Optional[str] = None,
+        transform: Optional[Callable] = None,
+        connection_factory: Optional[Callable[[str], Any]] = None,
+        worker_info_override=None,
+    ):
+        super().__init__()
+        if num_splits is None:
+            num_splits = world_size
+        if shuffle_seed is None:
+            shuffle_seed = random.randrange(2**32)
+        if num_splits % world_size != 0:
+            raise ValueError(
+                f"num_splits ({num_splits}) must be divisible by "
+                f"world_size ({world_size})"
+            )
+
+        self._table = table
+        self._num_splits = num_splits
+        self._shuffle = shuffle
+        self._shuffle_seed = shuffle_seed
+        self._epoch = epoch
+        self._rank = rank
+        self._world_size = world_size
+        self._read_batch_size = read_batch_size
+        self._prefetch_batches = prefetch_batches
+        self._columns = columns
+        self._shuffle_clump_size = shuffle_clump_size
+        self._filter = filter
+        self._transform = transform
+        self._connection_factory = connection_factory
+        self._worker_info_override = worker_info_override
+
+        # Live references to pipeline state, set only while __iter__ is running
+        # in the same process.  Used by the observability properties when the
+        # DataLoader runs with num_workers=0.
+        self._raw_batches_ref: Optional[list[deque]] = None
+        self._cooked_ref: Optional[list[deque]] = None
+        self._fetch_head_ref: Optional[list[int]] = None
+        self._split_sizes_ref: Optional[list[int]] = None
+        self._local_consumed_ref: Optional[list[int]] = None
+
+        # Shared-memory counters written by __iter__ (which may run in a
+        # DataLoader worker process) and read by the observability properties
+        # in the main process.  RawArray is picklable via the forkserver
+        # reduction protocol so it survives the dataset pickle round-trip.
+        # Layout: [unscanned_rows, raw_rows, cooked_rows, consumed_rows,
+        #          bytes_loaded, fetch_time_us, transform_time_us]
+        self._worker_stats: RawArray = RawArray(ctypes.c_int64, 7)
+
+        # Cumulative bytes of Arrow buffer data fetched across all iterations.
+        self._bytes_loaded: int = 0
+        # Cumulative seconds spent in LanceDB I/O and in transform functions.
+        self._fetch_time: float = 0.0
+        self._transform_time: float = 0.0
+
+        # Number of samples each split has already been consumed.  At global
+        # step boundaries all splits have consumed this many samples, so a
+        # single scalar captures the topology-independent checkpoint state.
+        self._resume_offset: int = 0
+
+        # Build the permutation table once, deterministically.
+        builder = permutation_builder(table)
+        if filter is not None:
+            builder = builder.filter(filter)
+        if shuffle:
+            perm_seed = shuffle_seed + epoch * _EPOCH_PRIME
+            self._perm_table = builder.split_random(
+                fixed=num_splits, seed=perm_seed, clump_size=shuffle_clump_size
+            ).execute()
+        else:
+            self._perm_table = builder.split_sequential(fixed=num_splits).execute()
+
+        # Contiguous block of global split indices assigned to this rank.
+        splits_per_rank = num_splits // world_size
+        rank_start = rank * splits_per_rank
+        self._rank_splits: list[int] = list(
+            range(rank_start, rank_start + splits_per_rank)
+        )
+
+    def _resolve_my_splits(self) -> list[int]:
+        """Return the split indices this instance should read in __iter__."""
+        torch_worker_info = get_worker_info()
+        if self._worker_info_override is not None:
+            if torch_worker_info is not None:
+                logger.warning(
+                    "worker_info_override is set but get_worker_info() also returned a "
+                    "non-None value; ignoring the real torch worker info and using the "
+                    "override instead.  This may lead to duplicated or incorrect data "
+                    "from the dataset."
+                )
+            worker_info = self._worker_info_override
+        else:
+            worker_info = torch_worker_info
+
+        if worker_info is None:
+            return self._rank_splits
+
+        num_workers: int = worker_info.num_workers
+        worker_id: int = worker_info.id
+        n_rank_splits = len(self._rank_splits)
+        if n_rank_splits % num_workers != 0:
+            raise ValueError(
+                f"Number of rank splits ({n_rank_splits}) must be divisible by "
+                f"num_workers ({num_workers})"
+            )
+        splits_per_worker = n_rank_splits // num_workers
+        start = worker_id * splits_per_worker
+        return self._rank_splits[start : start + splits_per_worker]
+
+    def __iter__(self) -> Iterator[dict[str, Any]]:
+        if self._raw_batches_ref is not None:
+            raise RuntimeError(
+                "StreamingDataset does not support concurrent iteration. "
+                "Only one active iterator per dataset instance is allowed."
+            )
+        my_splits = self._resolve_my_splits()
+        if not my_splits:
+            return
+
+        # Set identity transform on each Permutation so __getitems__ returns
+        # the raw RecordBatch.  Stage 2 applies the real transform.
+        permutations: list[Permutation] = []
+        for split_idx in my_splits:
+            perm = Permutation.from_tables(
+                self._table, self._perm_table, split=split_idx
+            )
+            if self._columns is not None:
+                perm = perm.select_columns(self._columns)
+            perm = perm.with_transform(lambda batch: batch)
+            if self._resume_offset > 0:
+                perm = perm.with_skip(self._resume_offset)
+            permutations.append(perm)
+
+        n = len(permutations)
+        split_sizes = [perm.num_rows for perm in permutations]
+        initial_offset = self._resume_offset
+        local_consumed = [0] * n
+
+        batch_size = self._read_batch_size
+        max_prefetch = self._prefetch_batches
+        cpu_workers = os.cpu_count() or 1
+        final_transform = (
+            self._transform if self._transform is not None else Transforms.arrow2python
+        )
+
+        # Per-split pipeline state.
+        fetch_head = [0] * n
+        io_pending = [deque() for _ in range(n)]  # Future[RecordBatch]
+        raw_batches = [deque() for _ in range(n)]  # RecordBatch — fetched, awaiting tx
+        tx_pending = [deque() for _ in range(n)]  # Future[list[Any]]
+        cooked = [deque() for _ in range(n)]  # rows ready to yield
+
+        # Limit simultaneous transforms to cpu_workers across all splits.
+        tx_semaphore = threading.Semaphore(cpu_workers)
+
+        # ── Stage 1 helpers ───────────────────────────────────────────────────
+
+        def _io_call(perm, indices):
+            t0 = time.perf_counter()
+            batch = perm.__getitems__(indices)
+            self._bytes_loaded += batch.nbytes
+            self._fetch_time += time.perf_counter() - t0
+            return batch
+
+        def _submit_io(i: int) -> None:
+            remaining = split_sizes[i] - fetch_head[i]
+            if remaining <= 0:
+                return
+            fetch = min(batch_size, remaining)
+            start = fetch_head[i]
+            fetch_head[i] += fetch
+            perm_i = permutations[i]
+            indices = list(range(start, start + fetch))
+            io_pending[i].append(io_pool.submit(_io_call, perm_i, indices))
+
+        def _fill_io(i: int) -> None:
+            while len(io_pending[i]) < max_prefetch and fetch_head[i] < split_sizes[i]:
+                _submit_io(i)
+
+        def _drain_io(i: int) -> None:
+            """Move completed I/O futures into raw_batches non-blockingly."""
+            while io_pending[i] and io_pending[i][0].done():
+                raw_batches[i].append(io_pending[i].popleft().result())
+
+        # ── Stage 2 helpers ───────────────────────────────────────────────────
+
+        def _tx_call_guarded(batch):
+            try:
+                t0 = time.perf_counter()
+                result = final_transform(batch)
+                self._transform_time += time.perf_counter() - t0
+                return result
+            finally:
+                tx_semaphore.release()
+
+        def _try_submit_tx(i: int) -> None:
+            """Submit transforms for raw_batches[i] up to available capacity."""
+            while raw_batches[i] and tx_semaphore.acquire(blocking=False):
+                batch = raw_batches[i].popleft()
+                tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+
+        def _drain_tx(i: int) -> None:
+            """Move completed transform futures into cooked non-blockingly."""
+            while tx_pending[i] and tx_pending[i][0].done():
+                cooked[i].extend(tx_pending[i].popleft().result())
+
+        # ── Combined advance ──────────────────────────────────────────────────
+
+        def _advance(i: int) -> None:
+            """Non-blocking pipeline pump for split i."""
+            _drain_io(i)
+            _drain_tx(i)
+            _try_submit_tx(i)
+            _fill_io(i)
+
+        def _ensure_cooked(i: int) -> None:
+            """Ensure cooked[i] has at least one row, blocking if necessary."""
+            _advance(i)
+            while not cooked[i]:
+                if tx_pending[i]:
+                    # Wait for the oldest in-flight transform.
+                    cooked[i].extend(tx_pending[i].popleft().result())
+                    _advance(i)
+                elif raw_batches[i]:
+                    # Acquire a transform slot (may block briefly if all
+                    # cpu_workers are busy with other splits).
+                    tx_semaphore.acquire()
+                    batch = raw_batches[i].popleft()
+                    tx_pending[i].append(tx_pool.submit(_tx_call_guarded, batch))
+                elif io_pending[i]:
+                    # Block on the oldest in-flight I/O fetch.
+                    raw_batches[i].append(io_pending[i].popleft().result())
+                    _advance(i)
+                else:
+                    break  # split exhausted
+
+        # ── Main loop ─────────────────────────────────────────────────────────
+
+        with ThreadPoolExecutor(max_workers=n * max_prefetch) as io_pool:
+            with ThreadPoolExecutor(max_workers=cpu_workers) as tx_pool:
+                self._raw_batches_ref = raw_batches
+                self._cooked_ref = cooked
+                self._fetch_head_ref = fetch_head
+                self._split_sizes_ref = split_sizes
+                self._local_consumed_ref = local_consumed
+                try:
+                    for i in range(n):
+                        _fill_io(i)
+
+                    while True:
+                        # Stop when any split is exhausted (all exhaust
+                        # simultaneously: equal split sizes + round-robin).
+                        if any(local_consumed[i] >= split_sizes[i] for i in range(n)):
+                            break
+
+                        for i in range(n):
+                            _ensure_cooked(i)
+                            row = cooked[i].popleft()
+                            local_consumed[i] += 1
+                            _advance(i)
+
+                            # After the last split in each cycle: update the
+                            # global offset and refresh the shared-memory stats
+                            # so the main process can observe pipeline depth
+                            # even when __iter__ runs in a worker process.
+                            if i == n - 1:
+                                self._resume_offset = initial_offset + local_consumed[i]
+                                ws = self._worker_stats
+                                ws[0] = sum(
+                                    split_sizes[j] - fetch_head[j] for j in range(n)
+                                )
+                                ws[1] = sum(
+                                    batch.num_rows for q in raw_batches for batch in q
+                                )
+                                ws[2] = sum(len(q) for q in cooked)
+                                ws[3] = sum(local_consumed)
+                                ws[4] = self._bytes_loaded
+                                ws[5] = int(self._fetch_time * 1_000_000)
+                                ws[6] = int(self._transform_time * 1_000_000)
+
+                            yield row
+                finally:
+                    self._raw_batches_ref = None
+                    self._cooked_ref = None
+                    self._fetch_head_ref = None
+                    self._split_sizes_ref = None
+                    self._local_consumed_ref = None
+
+    @property
+    def bytes_loaded(self) -> int:
+        """Cumulative bytes of raw Arrow buffer data fetched from storage.
+
+        Measured on the ``RecordBatch`` before any transform is applied, so
+        the value reflects actual I/O rather than the size of transformed
+        output.  Accumulates across multiple iterations of the same dataset
+        instance and is never reset automatically.
+        """
+        if self._raw_batches_ref is not None:
+            return self._bytes_loaded
+        return int(self._worker_stats[4])
+
+    @property
+    def fetch_time(self) -> float:
+        """Cumulative seconds spent waiting for data from LanceDB.
+
+        Measured per batch in the Stage 1 I/O threads as the total elapsed
+        time of the ``take_offsets`` call.  Accumulates across all splits and
+        all iterations.
+        """
+        if self._raw_batches_ref is not None:
+            return self._fetch_time
+        return self._worker_stats[5] / 1_000_000
+
+    @property
+    def transform_time(self) -> float:
+        """Cumulative seconds spent applying the transform.
+
+        Measured per batch in the Stage 2 transform threads as the elapsed
+        time inside the transform callable (or the default ``arrow2python``
+        conversion when no transform is set).  Accumulates across all splits
+        and all iterations.
+        """
+        if self._raw_batches_ref is not None:
+            return self._transform_time
+        return self._worker_stats[6] / 1_000_000
+
+    @property
+    def raw_queue_depth(self) -> int:
+        """Number of raw rows waiting for a transform thread across all splits.
+
+        A persistently non-zero value means Stage 2 (transform) is the
+        bottleneck: I/O is completing faster than transforms can consume
+        batches.  Returns 0 when not iterating.
+        """
+        if self._raw_batches_ref is not None:
+            return sum(batch.num_rows for q in self._raw_batches_ref for batch in q)
+        return int(self._worker_stats[1])
+
+    @property
+    def prefetch_queue_depth(self) -> int:
+        """Number of rows transformed and ready to yield across all splits.
+
+        Counts rows whose transform has completed and are sitting in memory
+        waiting for the main thread — rows that can be handed off with no
+        I/O or CPU wait.  Returns 0 when not iterating.
+        """
+        if self._cooked_ref is not None:
+            return sum(len(q) for q in self._cooked_ref)
+        return int(self._worker_stats[2])
+
+    @property
+    def unscanned_rows(self) -> int:
+        """Number of rows not yet submitted to the I/O stage across all splits.
+
+        Decreases as the I/O stage submits fetch requests.  When this reaches
+        zero all data has been requested from storage (though it may not have
+        arrived yet).  Returns 0 when not iterating.
+        """
+        if self._fetch_head_ref is not None:
+            return sum(
+                size - head
+                for size, head in zip(self._split_sizes_ref, self._fetch_head_ref)
+            )
+        return int(self._worker_stats[0])
+
+    @property
+    def consumed_rows(self) -> int:
+        """Number of rows already yielded to the caller across all splits.
+
+        Monotonically increases throughout iteration.  Returns 0 when not
+        iterating.
+        """
+        if self._local_consumed_ref is not None:
+            return sum(self._local_consumed_ref)
+        return int(self._worker_stats[3])
+
+    def __getstate__(self):
+        """Support pickling for multi-worker DataLoader (forkserver / spawn).
+
+        The live LanceDB table object contains non-picklable connection state
+        (sockets, Rust-backed PyO3 objects).  If a ``connection_factory`` was
+        supplied only the table name is serialised; the factory is called in
+        the worker to reopen the connection without embedding any credentials.
+        Without a factory the table's own picklable reopen state is captured
+        via ``_table_to_pickle_state`` (mirrors the ``Permutation`` approach).
+        """
+        state = self.__dict__.copy()
+        # _table: replace with reconnect info (credentials must not be embedded).
+        state["_table_name"] = self._table.name
+        if self._connection_factory is not None:
+            state["_table"] = None
+        else:
+            state["_table"] = _table_to_pickle_state(self._table)
+        # _perm_table: always in-memory; serialise as Arrow data (mirrors
+        # how Permutation.__getstate__ handles its permutation_table).
+        state["_perm_table"] = (
+            self._perm_table.name,
+            self._perm_table.to_arrow(),
+        )
+        for key in (
+            "_raw_batches_ref",
+            "_cooked_ref",
+            "_fetch_head_ref",
+            "_split_sizes_ref",
+            "_local_consumed_ref",
+        ):
+            state[key] = None
+        return state
+
+    def __setstate__(self, state):
+        """Reconnect to LanceDB after unpickling in a worker process."""
+        from . import connect as _connect
+
+        table_name = state.pop("_table_name")
+        table_state = state.pop("_table")
+        perm_name, perm_data = state.pop("_perm_table")
+        self.__dict__.update(state)
+        if self._connection_factory is not None:
+            self._table = self._connection_factory(table_name)
+        else:
+            self._table = _table_from_pickle_state(table_state)
+        self._perm_table = _connect("memory://").create_table(perm_name, perm_data)
+
+    def state_dict(self) -> dict:
+        """Snapshot the dataset's consumption state.
+
+        The returned dict is topology-independent: at global step boundaries
+        every split has been consumed the same number of times (by the
+        round-robin design), so the per-split count is a single uniform value
+        that is identical across all ranks and DataLoader workers.
+        """
+        return {
+            "shuffle_seed": self._shuffle_seed,
+            "num_splits": self._num_splits,
+            "epoch": self._epoch,
+            "samples_consumed_per_split": [self._resume_offset] * self._num_splits,
+        }
+
+    def load_state_dict(self, state: dict) -> None:
+        """Resume from a previously snapshotted state.
+
+        Raises ``ValueError`` if ``num_splits`` or ``shuffle_seed`` differ
+        from the checkpoint, since a different split structure or shuffle order
+        makes mid-epoch resumption meaningless.
+        """
+        if state["num_splits"] != self._num_splits:
+            raise ValueError(
+                f"num_splits mismatch: checkpoint has {state['num_splits']}, "
+                f"current dataset has {self._num_splits}"
+            )
+        if state["shuffle_seed"] != self._shuffle_seed:
+            raise ValueError(
+                f"shuffle_seed mismatch: checkpoint has {state['shuffle_seed']}, "
+                f"current dataset has {self._shuffle_seed}"
+            )
+        consumed = state["samples_consumed_per_split"]
+        # All entries are equal at step boundaries; use the first.
+        if isinstance(consumed, list):
+            self._resume_offset = consumed[0] if consumed else 0
+        else:
+            self._resume_offset = int(consumed)

python/python/lancedb/table.py

@@ -651,6 +651,16 @@ def _append_vector_columns(
                     col_data = func.compute_source_embeddings_with_retry(
                         batch[conf.source_column]
                     )
+                    # Replace vectors with wrong length (including empty lists
+                    # returned for inputs like empty strings) with None so that
+                    # _handle_bad_vectors can process them according to the
+                    # on_bad_vectors policy instead of crashing when PyArrow
+                    # tries to cast them into a fixed-size list array.
+                    expected_ndims = conf.function.ndims()
+                    col_data = [
+                        v if v is not None and len(v) == expected_ndims else None
+                        for v in col_data
+                    ]
                     if no_vector_column:
                         batch = batch.append_column(
                             schema.field(vector_column),
@@ -4020,7 +4030,16 @@ def _handle_bad_vector_column(
         dim = _infer_vector_dim(vec_arr)
         if dim is None:
             return data
-    has_wrong_dim = pc.not_equal(pc.list_value_length(vec_arr), dim)
+
+    is_null = pc.is_null(vec_arr)
+    # pc.list_value_length returns null for null list entries, so
+    # pc.not_equal(null, dim) also returns null. Use or_kleene so that
+    # True OR null = True (Kleene three-valued logic), ensuring null vectors
+    # are counted as wrong-dim.
+    has_wrong_dim = pc.or_kleene(
+        is_null,
+        pc.not_equal(pc.list_value_length(vec_arr), dim),
+    )
 
     has_bad_vectors = pc.any(has_nan).as_py() or pc.any(has_wrong_dim).as_py()
 

python/python/tests/test_embeddings.py

@@ -242,6 +242,49 @@ def test_embedding_with_bad_results(tmp_path):
     assert tbl["vector"].null_count == 1
 
 
+def test_embedding_with_empty_output_vectors(tmp_path):
+    """Regression test for issue #1672.
+
+    When an embedding function returns an empty list (e.g. for empty-string
+    inputs), _append_vector_columns used to crash because PyArrow cannot cast
+    [] into a fixed-size list element.  The fix replaces wrong-length vectors
+    with None before building the Arrow array so that _handle_bad_vectors can
+    process them normally.
+    """
+
+    @register("empty-vec-embedding")
+    class EmptyVecEmbeddingFunction(TextEmbeddingFunction):
+        def ndims(self):
+            return 128
+
+        def generate_embeddings(self, texts: Union[List[str], np.ndarray]) -> list:
+            # Simulate a model that returns an empty list for blank inputs
+            return [
+                [] if text.strip() == "" else np.random.randn(self.ndims()).tolist()
+                for text in texts
+            ]
+
+    db = lancedb.connect(tmp_path)
+    registry = EmbeddingFunctionRegistry.get_instance()
+    model = registry.get("empty-vec-embedding").create()
+
+    class Schema(LanceModel):
+        text: str = model.SourceField()
+        vector: Vector(model.ndims()) = model.VectorField()
+
+    table = db.create_table("test_empty_vec", schema=Schema, mode="overwrite")
+
+    # Should not crash; the row with the empty string should be dropped
+    table.add(
+        [{"text": "hello world"}, {"text": ""}, {"text": "foo"}],
+        on_bad_vectors="drop",
+    )
+
+    assert len(table) == 2
+    texts = table.to_arrow()["text"].to_pylist()
+    assert "" not in texts
+
+
 def test_with_existing_vectors(tmp_path):
     @register("mock-embedding")
     class MockEmbeddingFunction(TextEmbeddingFunction):

python/python/tests/test_expr.py

@@ -3,10 +3,14 @@
 
 """Tests for the type-safe expression builder API."""
 
-import pytest
+from datetime import date, datetime, timedelta, timezone
+from decimal import Decimal
+
 import pyarrow as pa
+import pytest
+
 import lancedb
-from lancedb.expr import Expr, col, lit, func
+from lancedb.expr import Expr, col, func, lit
 
 
 # ── unit tests for Expr construction ─────────────────────────────────────────
@@ -54,6 +58,28 @@ class TestExprConstruction:
         with pytest.raises(Exception):
             func("not_a_real_function", col("x"))
 
+    def test_lit_date(self):
+        e = lit(date(2024, 1, 1))
+        assert isinstance(e, Expr)
+
+    def test_lit_datetime(self):
+        # Naive datetime
+        e = lit(datetime(2024, 1, 1, 10, 0))
+        assert isinstance(e, Expr)
+
+    def test_lit_datetime_tz(self):
+        # Timezone-aware datetime
+        tz = timezone(timedelta(hours=5))
+        dt = datetime(2024, 1, 1, 10, 0, tzinfo=tz)
+        e = lit(dt)
+        assert isinstance(e, Expr)
+
+    def test_lit_decimal_precision(self):
+        # High precision Decimal that would be rounded if converted to float
+        d = Decimal("1.234567890123456789")
+        e = lit(d)
+        assert isinstance(e, Expr)
+
 
 class TestExprOperators:
     def test_eq_operator(self):
@@ -142,6 +168,20 @@ class TestExprOperators:
         assert isinstance(e, Expr)
         assert e.to_sql() == "(name = 'alice')"
 
+    def test_reflexive_comparisons(self):
+        # 10 < col("age") swaps to col("age") > 10
+        assert (10 < col("age")).to_sql() == "(age > 10)"
+        assert (10 <= col("age")).to_sql() == "(age >= 10)"
+        assert (10 > col("age")).to_sql() == "(age < 10)"
+        assert (10 >= col("age")).to_sql() == "(age <= 10)"
+        assert (10 == col("age")).to_sql() == "(age = 10)"
+        assert (10 != col("age")).to_sql() == "(age <> 10)"
+
+    def test_reflexive_logical(self):
+        # True & Expr calls Expr.__rand__(True)
+        assert (True & (col("age") > 18)).to_sql() == "(true AND (age > 18))"
+        assert (False | (col("age") > 18)).to_sql() == "(false OR (age > 18))"
+
 
 class TestExprBytesLiteral:
     def test_bytes_to_sql(self):
@@ -282,6 +322,40 @@ class TestExprRepr:
             {e: 1}
 
 
+class TestExprReflexive:
+    def test_reflexive_eq(self):
+        e = 1 == col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x = 1)"
+
+    def test_reflexive_ne(self):
+        e = 1 != col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x <> 1)"
+
+    def test_reflexive_lt(self):
+        # 1 < x  =>  (x > 1)
+        e = 1 < col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x > 1)"
+
+    def test_reflexive_gt(self):
+        # 1 > x  =>  (x < 1)
+        e = 1 > col("x")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(x < 1)"
+
+    def test_reflexive_and(self):
+        e = True & col("active")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(true AND active)"
+
+    def test_reflexive_or(self):
+        e = False | col("inactive")
+        assert isinstance(e, Expr)
+        assert e.to_sql() == "(false OR inactive)"
+
+
 # ── integration tests: end-to-end query against a real table ─────────────────
 
 
@@ -432,6 +506,72 @@ class TestColNamingIntegration:
         assert sorted(result["upper_name"].to_pylist()) == ["ALICE", "BOB", "CHARLIE"]
 
 
+@pytest.fixture
+def type_check_table(tmp_path):
+    """Fixture that creates a table with Date32 and Decimal128 columns."""
+    db = lancedb.connect(str(tmp_path))
+    schema = pa.schema(
+        [
+            ("date", pa.date32()),
+            ("decimal", pa.decimal128(10, 2)),
+            ("binary", pa.binary()),
+        ]
+    )
+    data = pa.table(
+        {
+            "date": [date(2024, 1, 1), date(2024, 1, 2)],
+            "decimal": [Decimal("10.50"), Decimal("20.75")],
+            "binary": [b"\x01", b"\x02"],
+        },
+        schema=schema,
+    )
+    return db.create_table("extended_types", data)
+
+
+class TestExtendedTypeIntegration:
+    """Integration tests verifying that typed literals work correctly in filters."""
+
+    def test_date_integration(self, type_check_table):
+        """Verify that Date32 literals are correctly parsed and filtered."""
+        result = (
+            type_check_table.search()
+            .where(col("date") == lit(date(2024, 1, 1)))
+            .to_arrow()
+        )
+        assert result.num_rows == 1
+        assert result["date"][0].as_py() == date(2024, 1, 1)
+
+    def test_decimal_integration(self, tmp_path):
+        """A Decimal literal must retain full 128-bit precision in a filter.
+
+        1.234567890123456789 and 1.234567890123456790 differ only in the last
+        digit and are indistinguishable once truncated to f64.  The filter
+        therefore returns the single expected row only if ``lit(Decimal)``
+        produces a true ``Decimal128`` scalar rather than being coerced to f64.
+        """
+        low = Decimal("1.234567890123456789")
+        high = Decimal("1.234567890123456790")
+
+        db = lancedb.connect(str(tmp_path / "decimal_precision"))
+        schema = pa.schema([("val", pa.decimal128(19, 18))])
+        table = db.create_table(
+            "decimal_precision",
+            pa.table({"val": [low, high]}, schema=schema),
+        )
+
+        result = table.search().where(col("val") < lit(high)).to_arrow()
+        assert result.num_rows == 1
+        assert result["val"][0].as_py() == low
+
+    def test_binary_integration(self, type_check_table):
+        """Verify that Binary literals are correctly filtered."""
+        result = (
+            type_check_table.search().where(col("binary") == lit(b"\x01")).to_arrow()
+        )
+        assert result.num_rows == 1
+        assert result["binary"][0].as_py() == b"\x01"
+
+
 # ── bytes / binary column integration tests ───────────────────────────────────
 
 

python/python/tests/test_remote_db.py

@@ -412,10 +412,12 @@ def test_remote_permutation_is_picklable():
             content_len = int(request.headers.get("Content-Length"))
             body = json.loads(request.rfile.read(content_len))
             if "filter" in body:
-                match = re.search(r"_rowoffset in \((.*?)\)", body["filter"])
+                match = re.search(
-                offsets = [int(offset.strip()) for offset in match.group(1).split(",")]
+                    r"_rowoffset\s+in\s+\((.*?)\)", body["filter"], re.IGNORECASE
+                )
+                offsets = [int(o.strip()) for o in match.group(1).split(",")]
             else:
-                offsets = rows
+                offsets = list(range(len(rows)))
             table = pa.table({"a": [rows[offset] for offset in offsets]})
 
             request.send_response(200)

python/src/expr.rs

@@ -7,12 +7,14 @@
 //! build type-safe filter / projection expressions that map directly to
 //! DataFusion [`Expr`] nodes, bypassing SQL string parsing.
 
+use std::ops::{Add, Div, Mul, Not, Sub};
+
 use arrow::{datatypes::DataType, pyarrow::PyArrowType};
 use datafusion_common::ScalarValue;
 use lancedb::expr::{
     DfExpr, col as ldb_col, contains, expr_cast, is_in, lit as df_lit, lower, upper,
 };
-use pyo3::types::PyBytes;
+use pyo3::types::{PyBytes, PyDate, PyDateTime};
 use pyo3::{Bound, PyAny, PyResult, exceptions::PyValueError, prelude::*, pyfunction};
 
 /// A type-safe DataFusion expression.
@@ -63,30 +65,30 @@ impl PyExpr {
         Self(self.0.clone().or(other.0.clone()))
     }
 
+    /// Logical NOT.
     fn not_(&self) -> Self {
-        use std::ops::Not;
         Self(self.0.clone().not())
     }
 
     // ── arithmetic ───────────────────────────────────────────────────────────
 
+    /// Add expressions.
     fn add(&self, other: &Self) -> Self {
-        use std::ops::Add;
         Self(self.0.clone().add(other.0.clone()))
     }
 
+    /// Subtract expressions.
     fn sub(&self, other: &Self) -> Self {
-        use std::ops::Sub;
         Self(self.0.clone().sub(other.0.clone()))
     }
 
+    /// Multiply expressions.
     fn mul(&self, other: &Self) -> Self {
-        use std::ops::Mul;
         Self(self.0.clone().mul(other.0.clone()))
     }
 
+    /// Divide expressions.
     fn div(&self, other: &Self) -> Self {
-        use std::ops::Div;
         Self(self.0.clone().div(other.0.clone()))
     }
 
@@ -153,7 +155,8 @@ pub fn expr_col(name: &str) -> PyExpr {
 
 /// Create a literal value expression.
 ///
-/// Supported Python types: `bool`, `int`, `float`, `str`, `bytes`.
+/// Supported Python types: `bool`, `int`, `float`, `str`, `bytes`, `date`,
+/// `datetime`, `Decimal`.
 #[pyfunction]
 pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
     // bool must be checked before int because bool is a subclass of int in Python
@@ -163,6 +166,19 @@ pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
     if let Ok(i) = value.extract::<i64>() {
         return Ok(PyExpr(df_lit(i)));
     }
+    // Decimal must be checked before f64: Python's Decimal implements __float__,
+    // so value.extract::<f64>() would succeed and silently truncate the value to
+    // f64, losing precision. Build a Decimal128 scalar to preserve it instead.
+    if value.get_type().name()? == "Decimal" {
+        let s = value.call_method0("__str__")?.extract::<String>()?;
+        // Parse the decimal string into an i128 value, precision, and scale.
+        let (val, precision, scale) = parse_decimal(&s)?;
+        return Ok(PyExpr(df_lit(ScalarValue::Decimal128(
+            Some(val),
+            precision,
+            scale,
+        ))));
+    }
     if let Ok(f) = value.extract::<f64>() {
         return Ok(PyExpr(df_lit(f)));
     }
@@ -173,12 +189,48 @@ pub fn expr_lit(value: Bound<'_, PyAny>) -> PyResult<PyExpr> {
         let bytes = value.extract::<Vec<u8>>()?;
         return Ok(PyExpr(df_lit(ScalarValue::Binary(Some(bytes)))));
     }
+
+    // datetime.datetime is a subclass of datetime.date, so it must be checked first.
+    if let Ok(dt) = value.cast::<PyDateTime>() {
+        let ts: f64 = dt.call_method0("timestamp")?.extract()?;
+        let micros = (ts * 1_000_000.0).round() as i64;
+        return Ok(PyExpr(df_lit(ScalarValue::TimestampMicrosecond(
+            Some(micros),
+            None,
+        ))));
+    }
+    if let Ok(d) = value.cast::<PyDate>() {
+        let ordinal: i32 = d.call_method0("toordinal")?.extract()?;
+        let days = ordinal - 719163; // Unix epoch is 1970-01-01
+        return Ok(PyExpr(df_lit(ScalarValue::Date32(Some(days)))));
+    }
+
     Err(PyValueError::new_err(format!(
-        "unsupported literal type: {}. Supported: bool, int, float, str, bytes",
+        "unsupported literal type: {}. Supported: bool, int, float, str, bytes, date, datetime, Decimal",
         value.get_type().name()?
     )))
 }
 
+fn parse_decimal(s: &str) -> PyResult<(i128, u8, i8)> {
+    let s = s.trim();
+    let dot_pos = s.find('.');
+    let scale = if let Some(pos) = dot_pos {
+        (s.len() - pos - 1) as i8
+    } else {
+        0
+    };
+
+    let digits = s.replace('.', "");
+    let val = digits
+        .parse::<i128>()
+        .map_err(|e| PyValueError::new_err(format!("failed to parse decimal digits: {}", e)))?;
+
+    // Precision is total number of digits
+    let precision = digits.trim_start_matches('-').len() as u8;
+
+    Ok((val, precision, scale))
+}
+
 /// Call an arbitrary registered SQL function by name.
 ///
 /// See `lancedb::expr::func` for the list of supported function names.

python/src/permutation.rs

@@ -79,13 +79,14 @@ impl PyAsyncPermutationBuilder {
 
 #[pymethods]
 impl PyAsyncPermutationBuilder {
-    #[pyo3(signature = (*, ratios=None, counts=None, fixed=None, seed=None, split_names=None))]
+    #[pyo3(signature = (*, ratios=None, counts=None, fixed=None, seed=None, clump_size=None, split_names=None))]
     pub fn split_random(
         slf: PyRefMut<'_, Self>,
         ratios: Option<Vec<f64>>,
         counts: Option<Vec<u64>>,
         fixed: Option<u64>,
         seed: Option<u64>,
+        clump_size: Option<u64>,
         split_names: Option<Vec<String>>,
     ) -> PyResult<Self> {
         // Check that exactly one split type is provided
@@ -111,7 +112,14 @@ impl PyAsyncPermutationBuilder {
         };
 
         slf.modify(|builder| {
-            builder.with_split_strategy(SplitStrategy::Random { seed, sizes }, split_names)
+            builder.with_split_strategy(
+                SplitStrategy::Random {
+                    seed,
+                    sizes,
+                    clump_size,
+                },
+                split_names,
+            )
         })
     }
 

rust/lancedb/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.31.0-beta.5"
+version = "0.31.0-beta.6"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true
@@ -108,6 +108,8 @@ http-body = "1"                                        # Matching reqwest
 rstest = "0.23.0"
 test-log = "0.2"
 serial_test = "3"
+[target.'cfg(unix)'.dev-dependencies]
+pprof = { version = "0.14", features = ["flamegraph"] }
 
 
 [features]
@@ -164,6 +166,9 @@ required-features = ["sentence-transformers"]
 name = "bedrock"
 required-features = ["bedrock"]
 
+[[example]]
+name = "bench_streaming_dataloader"
+
 [[example]]
 name = "simple"
 

rust/lancedb/examples/bench_streaming_dataloader.rs

@@ -0,0 +1,272 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+//! Benchmark + CPU profiler for the PermutationReader used by the elastic
+//! streaming dataloader.
+//!
+//! Normal sweep:
+//!   cargo run --release --example bench_streaming_dataloader
+//!
+//! Flamegraph (self-contained, no perf/dtrace needed):
+//!   BENCH_PROFILE=1 BENCH_CHUNK=64 cargo run --release \
+//!       --example bench_streaming_dataloader
+//!   # writes flamegraph.svg in the current directory
+//!
+//! Environment variables:
+//!   BENCH_NUM_ROWS   – total rows (default 49152 = 24 × 2048)
+//!   BENCH_NUM_SPLITS – number of splits (default 24)
+//!   BENCH_STEPS      – round-robin cycles per chunk-size trial (default 200)
+//!   BENCH_ROW_BYTES  – bytes of payload per row (default 4096)
+//!   BENCH_CHUNK      – restrict sweep to this single chunk size
+//!   BENCH_PROFILE    – if set to "1", capture a pprof flamegraph SVG
+
+use std::{sync::Arc, time::Instant};
+
+use arrow_array::{Int32Array, LargeBinaryArray, RecordBatch};
+use arrow_schema::{DataType, Field, Schema};
+use lancedb::{
+    Result, Table,
+    arrow::{SendableRecordBatchStream, SimpleRecordBatchStream},
+    connect,
+    dataloader::permutation::{
+        builder::{PermutationBuilder, ShuffleStrategy},
+        reader::PermutationReader,
+        split::{SplitSizes, SplitStrategy},
+    },
+    query::Select,
+};
+
+fn env_usize(key: &str, default: usize) -> usize {
+    std::env::var(key)
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(default)
+}
+
+// ---------------------------------------------------------------------------
+// Table creation
+// ---------------------------------------------------------------------------
+
+async fn make_base_table(num_rows: usize, row_bytes: usize) -> Result<Table> {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Int32, false),
+        Field::new("payload", DataType::LargeBinary, false),
+    ]));
+    let payload = vec![0u8; row_bytes];
+    let ids: Int32Array = (0..num_rows as i32).collect();
+    let payloads: LargeBinaryArray = (0..num_rows).map(|_| Some(payload.as_slice())).collect();
+    let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids), Arc::new(payloads)])?;
+    let stream: SendableRecordBatchStream = Box::pin(SimpleRecordBatchStream::new(
+        futures::stream::once(std::future::ready(Ok(batch))),
+        schema,
+    ));
+    let db = connect("memory:///").execute().await?;
+    db.create_table("base", stream).execute().await
+}
+
+async fn make_permutation_table(base: &Table, num_splits: usize) -> Result<Table> {
+    PermutationBuilder::new(base.clone())
+        .with_split_strategy(
+            SplitStrategy::Random {
+                seed: Some(42),
+                sizes: SplitSizes::Fixed(num_splits as u64),
+                clump_size: None,
+            },
+            None,
+        )
+        .with_shuffle_strategy(ShuffleStrategy::Random {
+            seed: Some(42),
+            clump_size: None,
+        })
+        .build()
+        .await
+}
+
+// ---------------------------------------------------------------------------
+// Round-robin hot loop (mirrors StreamingDataset.__iter__)
+// ---------------------------------------------------------------------------
+
+async fn run_hot_loop(
+    readers: &[PermutationReader],
+    chunk_size: usize,
+    steps: usize,
+) -> Result<(usize, f64)> {
+    let n = readers.len();
+    let split_sizes: Vec<usize> = readers.iter().map(|r| r.count_rows() as usize).collect();
+
+    struct SplitBuf {
+        batch: Option<RecordBatch>,
+        row_in_batch: usize,
+        consumed: usize,
+    }
+    let mut bufs: Vec<SplitBuf> = (0..n)
+        .map(|_| SplitBuf {
+            batch: None,
+            row_in_batch: 0,
+            consumed: 0,
+        })
+        .collect();
+
+    // Pre-fill
+    for i in 0..n {
+        let fetch = chunk_size.min(split_sizes[i]);
+        if fetch > 0 {
+            let offsets: Vec<u64> = (0..fetch as u64).collect();
+            bufs[i].batch = Some(readers[i].take_offsets(&offsets, Select::All).await?);
+        }
+    }
+
+    let mut total_rows = 0usize;
+    let t0 = Instant::now();
+
+    'outer: for _step in 0..steps {
+        for i in 0..n {
+            if bufs[i].consumed >= split_sizes[i] {
+                break 'outer;
+            }
+            let need_refill = bufs[i]
+                .batch
+                .as_ref()
+                .map(|b| bufs[i].row_in_batch >= b.num_rows())
+                .unwrap_or(true);
+            if need_refill {
+                let start = bufs[i].consumed as u64;
+                let remaining = (split_sizes[i] - bufs[i].consumed) as u64;
+                let fetch = chunk_size.min(remaining as usize);
+                let offsets: Vec<u64> = (start..start + fetch as u64).collect();
+                bufs[i].batch = Some(readers[i].take_offsets(&offsets, Select::All).await?);
+                bufs[i].row_in_batch = 0;
+            }
+            bufs[i].row_in_batch += 1;
+            bufs[i].consumed += 1;
+            total_rows += 1;
+        }
+    }
+
+    Ok((total_rows, t0.elapsed().as_secs_f64()))
+}
+
+// ---------------------------------------------------------------------------
+// Main
+// ---------------------------------------------------------------------------
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let num_splits = env_usize("BENCH_NUM_SPLITS", 24);
+    let num_rows = env_usize("BENCH_NUM_ROWS", num_splits * 2048);
+    let steps = env_usize("BENCH_STEPS", 200);
+    let row_bytes = env_usize("BENCH_ROW_BYTES", 4096);
+    let single_chunk: Option<usize> = std::env::var("BENCH_CHUNK")
+        .ok()
+        .and_then(|v| v.parse().ok());
+    let do_profile = std::env::var("BENCH_PROFILE")
+        .map(|v| v == "1")
+        .unwrap_or(false);
+
+    assert_eq!(
+        num_rows % num_splits,
+        0,
+        "NUM_ROWS must be divisible by NUM_SPLITS"
+    );
+
+    println!("Benchmark config:");
+    println!(
+        "  num_rows={}  num_splits={}  rows/split={}  steps={}  row_bytes={}",
+        num_rows,
+        num_splits,
+        num_rows / num_splits,
+        steps,
+        row_bytes,
+    );
+    println!(
+        "  ~{:.1} MB total",
+        (num_rows * row_bytes) as f64 / (1024.0 * 1024.0)
+    );
+    println!();
+
+    print!("Building base table... ");
+    let _ = std::io::Write::flush(&mut std::io::stdout());
+    let base = make_base_table(num_rows, row_bytes).await?;
+    println!("done");
+
+    print!("Building permutation table... ");
+    let _ = std::io::Write::flush(&mut std::io::stdout());
+    let perm = make_permutation_table(&base, num_splits).await?;
+    println!("done");
+
+    print!("Building {} PermutationReaders... ", num_splits);
+    let _ = std::io::Write::flush(&mut std::io::stdout());
+    let base_inner = base.base_table().clone();
+    let perm_inner = perm.base_table().clone();
+    let mut readers = Vec::with_capacity(num_splits);
+    for split in 0..num_splits {
+        readers.push(
+            PermutationReader::try_from_tables(
+                base_inner.clone(),
+                perm_inner.clone(),
+                split as u64,
+            )
+            .await?,
+        );
+    }
+    println!("done ({} rows/split)", readers[0].count_rows());
+    println!();
+
+    let chunk_sizes: Vec<usize> = if let Some(c) = single_chunk {
+        vec![c]
+    } else {
+        vec![1, 4, 16, 64, 256, 1024, 4096, 16384]
+    };
+
+    if do_profile {
+        #[cfg(unix)]
+        {
+            let chunk = chunk_sizes[0];
+            println!("Profiling chunk={chunk} for {steps} steps...");
+            // Warm-up outside the profiler window
+            let _ = run_hot_loop(&readers, chunk, 1).await?;
+
+            let guard = pprof::ProfilerGuardBuilder::default()
+                .frequency(1000)
+                .build()
+                .unwrap();
+
+            let (rows, elapsed) = run_hot_loop(&readers, chunk, steps).await?;
+
+            if let Ok(report) = guard.report().build() {
+                let svg_path = "flamegraph.svg";
+                let file = std::fs::File::create(svg_path).unwrap();
+                report.flamegraph(file).unwrap();
+                println!("Flamegraph written to {svg_path}");
+            }
+
+            let rows_per_sec = rows as f64 / elapsed;
+            println!("chunk={chunk}  {rows} rows  {elapsed:.3}s  {rows_per_sec:.0} rows/s");
+        }
+        #[cfg(not(unix))]
+        {
+            println!("Flamegraph profiling (BENCH_PROFILE=1) is not supported on this platform.");
+            println!("Run without BENCH_PROFILE to get throughput numbers.");
+        }
+    } else {
+        println!(
+            "{:>6}  {:>7}  {:>8}  {:>11}  {:>10}",
+            "chunk", "rows", "elapsed", "rows/s", "ms/step"
+        );
+        println!("{}", "-".repeat(52));
+
+        for &chunk in &chunk_sizes {
+            let _ = run_hot_loop(&readers, chunk, 1).await?;
+            let (rows, elapsed) = run_hot_loop(&readers, chunk, steps).await?;
+            let rows_per_sec = rows as f64 / elapsed;
+            let ms_per_step = elapsed / steps as f64 * 1000.0;
+            println!(
+                "{:>6}  {:>7}  {:>7.3}s  {:>11.0}  {:>9.1}ms",
+                chunk, rows, elapsed, rows_per_sec, ms_per_step,
+            );
+        }
+    }
+
+    println!("\nDone.");
+    Ok(())
+}

rust/lancedb/src/dataloader/permutation/builder.rs

@@ -391,6 +391,7 @@ mod tests {
                 SplitStrategy::Random {
                     seed: Some(42),
                     sizes: SplitSizes::Percentages(vec![0.05, 0.30]),
+                    clump_size: None,
                 },
                 None,
             )

rust/lancedb/src/dataloader/permutation/reader.rs

@@ -21,6 +21,7 @@ use arrow::compute::concat_batches;
 use arrow::datatypes::UInt64Type;
 use arrow_array::{RecordBatch, UInt64Array};
 use arrow_schema::SchemaRef;
+use datafusion_expr::{Expr, col, lit};
 use futures::{StreamExt, TryStreamExt};
 use lance::dataset::scanner::DatasetRecordBatchStream;
 use lance::io::RecordBatchStream;
@@ -196,17 +197,10 @@ impl PermutationReader {
             .expect_ok()?
             .values();
 
-        let filter = format!(
+        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();
-            "_rowid in ({})",
-            row_ids
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
 
         let base_query = QueryRequest {
-            filter: Some(QueryFilter::Sql(filter)),
+            filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
             select: selection,
             with_row_id: true,
             ..Default::default()

rust/lancedb/src/dataloader/permutation/split.rs

@@ -35,9 +35,13 @@ pub enum SplitStrategy {
     /// Rows will be randomly assigned to splits
     ///
     /// A seed can be provided to make the assignment deterministic.
+    ///
+    /// A clump_size can be provided to shuffle contiguous groups of rows together,
+    /// preserving I/O locality while still randomising the split assignment.
     Random {
         seed: Option<u64>,
         sizes: SplitSizes,
+        clump_size: Option<u64>,
     },
     /// Rows will be assigned to splits based on the values in the specified columns.
     ///
@@ -323,13 +327,17 @@ impl Splitter {
                 self.apply_sequential(source, num_rows, &SplitSizes::Counts(vec![num_rows]))
                     .await
             }
-            SplitStrategy::Random { seed, sizes } => {
+            SplitStrategy::Random {
+                seed,
+                sizes,
+                clump_size,
+            } => {
                 let shuffler = Shuffler::new(ShufflerConfig {
                     seed: *seed,
                     // In this case we are only shuffling row ids so we can use a large max_rows_per_file
                     max_rows_per_file: 10 * 1024 * 1024,
                     temp_dir: self.temp_dir.clone(),
-                    clump_size: None,
+                    clump_size: *clump_size,
                 });
 
                 let shuffled = shuffler.shuffle(source, num_rows).await?;
@@ -692,6 +700,7 @@ mod tests {
             SplitStrategy::Random {
                 seed: Some(42),
                 sizes: SplitSizes::Fixed(3),
+                clump_size: None,
             },
         );
 
@@ -718,6 +727,7 @@ mod tests {
             SplitStrategy::Random {
                 seed: Some(42),
                 sizes: SplitSizes::Counts(vec![5, 15, 10]),
+                clump_size: None,
             },
         );
 
@@ -744,6 +754,7 @@ mod tests {
             SplitStrategy::Random {
                 seed: Some(42),
                 sizes: SplitSizes::Percentages(vec![0.217, 0.168, 0.17]),
+                clump_size: None,
             },
         );
 

rust/lancedb/src/query.rs

@@ -7,7 +7,7 @@ use std::{future::Future, time::Duration};
 use arrow::compute::concat_batches;
 use arrow_array::{Array, Float16Array, Float32Array, Float64Array, RecordBatch, make_array};
 use arrow_schema::{DataType, SchemaRef};
-use datafusion_expr::Expr;
+use datafusion_expr::{Expr, col, lit};
 use datafusion_physical_plan::ExecutionPlan;
 use futures::{FutureExt, TryFutureExt, TryStreamExt, stream, try_join};
 use half::f16;
@@ -1468,18 +1468,13 @@ impl TakeQuery {
     ///
     /// See [`crate::Table::take_offsets`] for more details.
     pub fn from_offsets(parent: Arc<dyn BaseTable>, offsets: Vec<u64>) -> Self {
-        let filter = format!(
+        let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
-            "_rowoffset in ({})",
-            offsets
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
         Self {
             parent,
             request: QueryRequest {
-                filter: Some(QueryFilter::Sql(filter)),
+                filter: Some(QueryFilter::Datafusion(
+                    col("_rowoffset").in_list(in_list, false),
+                )),
                 ..Default::default()
             },
         }
@@ -1489,18 +1484,11 @@ impl TakeQuery {
     ///
     /// See [`crate::Table::take_row_ids`] for more details.
     pub fn from_row_ids(parent: Arc<dyn BaseTable>, row_ids: Vec<u64>) -> Self {
-        let filter = format!(
+        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();
-            "_rowid in ({})",
-            row_ids
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
         Self {
             parent,
             request: QueryRequest {
-                filter: Some(QueryFilter::Sql(filter)),
+                filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
                 ..Default::default()
             },
         }