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

feat: support date, datetime, bytes, and Decimal literals in expr builder (#3235 )
### **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>
2026-08-27 16:38:31 +00:00 · 2026-07-06 21:10:10 +00:00 · 2026-07-06 11:30:16 -07:00 · 2026-07-06 05:50:45 -07:00 · 2026-07-04 11:11:10 -07:00 · 2026-07-02 16:54:34 -07:00
37 changed files with 3163 additions and 133 deletions
@@ -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*)\\.
@@ -3423,8 +3423,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"

 [[package]]
 name = "fsst"
-version = "9.0.0-beta.10"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
-source = "git+https://github.com/lance-format/lance.git?tag=v9.0.0-beta.10#e25b71e74b89d10c57b412d111bde087117383f3"
+version = "9.0.0-beta.16"
+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"
@@ -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-core = { "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.10", "tag" = "v9.0.0-beta.10", "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-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-index = { "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.10", "tag" = "v9.0.0-beta.10", "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-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-table = { "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.10", "tag" = "v9.0.0-beta.10", "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-encoding = { "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.10", "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.16", "tag" = "v9.0.0-beta.16", "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.16", "tag" = "v9.0.0-beta.16", "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.16", "tag" = "v9.0.0-beta.16", "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.16", "tag" = "v9.0.0-beta.16", "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.16", "tag" = "v9.0.0-beta.16", "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.16", "tag" = "v9.0.0-beta.16", "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.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 }
@@ -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" },
 ]

 # ---------------------------------------------------------------------------
@@ -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>
 ```

@@ -8,6 +8,14 @@

 ## Properties

+### clumpSize?
+
+```ts
+optional clumpSize: number;
+```
+
+***
+
 ### counts?

 ```ts
@@ -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>

@@ -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>
@@ -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
@@ -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",
@@ -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",
@@ -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",
@@ -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",
@@ -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",
@@ -1,6 +1,6 @@
 {
  "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.31.0-beta.5",
+  "version": "0.31.0-beta.6",
  "os": [
    "win32"
  ],
@@ -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",
@@ -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"
@@ -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",
@@ -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(),
            )
        })
@@ -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()
@@ -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:
@@ -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))


@@ -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
@@ -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)
@@ -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()

@@ -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):
@@ -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 ───────────────────────────────────


@@ -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"])
-                offsets = [int(offset.strip()) for offset in match.group(1).split(",")]
+                match = re.search(
+                    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)
@@ -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.
@@ -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,
+            )
        })
    }

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

@@ -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(())
+}
@@ -391,6 +391,7 @@ mod tests {
                SplitStrategy::Random {
                    seed: Some(42),
                    sizes: SplitSizes::Percentages(vec![0.05, 0.30]),
+                    clump_size: None,
                },
                None,
            )
@@ -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!(
-            "_rowid in ({})",
-            row_ids
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
+        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();

        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()
@@ -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,
            },
        );

@@ -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!(
-            "_rowoffset in ({})",
-            offsets
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
+        let in_list: Vec<Expr> = offsets.iter().map(|o| lit(*o)).collect();
        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!(
-            "_rowid in ({})",
-            row_ids
-                .iter()
-                .map(|o| o.to_string())
-                .collect::<Vec<_>>()
-                .join(",")
-        );
+        let in_list: Vec<Expr> = row_ids.iter().map(|id| lit(*id)).collect();
        Self {
            parent,
            request: QueryRequest {
-                filter: Some(QueryFilter::Sql(filter)),
+                filter: Some(QueryFilter::Datafusion(col(ROW_ID).in_list(in_list, false))),
                ..Default::default()
            },
        }
Author	SHA1	Message	Date
lancedb automation	5e58dde32d	chore: update lance dependency to v9.0.0-beta.16	2026-07-06 21:10:10 +00:00
LAKSH JAIN	3bcff0165e	feat: support date, datetime, bytes, and Decimal literals in expr builder (#3235 ) ### 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>	2026-07-06 11:30:16 -07:00
Weston Pace	c6db80dd0b	feat: add an elastic dataloader as an iterable dataset (#3509 ) # Elastic Streaming Dataloader ## Motivation Training large models on LanceDB tables today requires loading the entire dataset into memory or writing bespoke batching logic. This PR introduces `StreamingDataset`, a PyTorch `IterableDataset` that streams directly from a LanceDB table with two hard guarantees that are difficult to achieve together: elastic determinism and resumability. ## Goals ### Elastic determinism The dataset partitions the table into a fixed number of splits (controlled by `num_splits`, `shuffle_seed`, and `epoch`). Samples are yielded by round-robining over splits one sample per split per cycle. Because the split structure is fixed, the set of samples that makes up each global training step is identical regardless of `world_size` or `num_workers`. You can scale your cluster up or down between runs and the model sees the same data in the same order — no re-sharding, no gradient variance from topology changes. ### Resumability `state_dict()` / `load_state_dict()` capture how many samples each split has consumed. Because all splits are the same size and the round-robin design keeps them in lockstep, the state reduces to a single scalar (`samples_consumed_per_split`) that is topology-independent. A checkpoint saved with 8 GPUs can resume correctly on 4 GPUs or 16 GPUs without any adjustment. ### PyTorch `IterableDataset` / streaming `StreamingDataset` implements the standard PyTorch `IterableDataset` interface, so it drops into any existing `DataLoader` pipeline without modification. Data is fetched lazily from Lance in chunks — only the rows needed for the current batch are ever in memory. Compared to the map dataset this takes more work from pytorch and puts it into the dataset itself (e.g. shuffling, filtering, etc.). We do this because we cannot achieve things like elastic determinism or prefiltering otherwise. ### Multi-worker support DataLoader workers are automatically assigned contiguous sub-blocks of splits (the rank's splits are divided evenly across workers). Each worker is independent: no shared state, no inter-process coordination. The only constraint is that `num_splits` must be divisible by `world_size * num_workers`. That being said, multi-worker is highly discouraged as it relies on multiprocessing which is inefficient. Still, we want to support it. ### Filters as prefilters Filters are applied at permutation-build time via `PermutationBuilder.filter()`, not re-evaluated on every fetch. The filtered row IDs are stored in the permutation table so that subsequent reads see only the matching rows. This allows us to avoid loading rows that don't match the filter (which is the default pytorch behavior) ### Prefetching Two parameters control the I/O pipeline: - `read_batch_size` (default 64) — number of rows fetched per `take_offsets` call. Larger values amortise per-request overhead, which is critical on object storage where a single round-trip can cost ~100 ms. - `prefetch_batches` (default 4) — number of batches prefetched in parallel per split via a `ThreadPoolExecutor`. While the model processes the current batch, the next several batches are already in flight, hiding storage latency behind compute. If set correctly then you can get good performance even with num_workers=0 (unless you are bottlenecked on transform). ### Transform parallelism The underlying `Permutation` API supports a `with_transform()` callback for decoding, augmentation, and format conversion. Unfortunately, this is not parallelized. Pytorch typically parallelizes this with num_workers which is multiprocessing which is highly inefficient. For simple transforms we should be able to utilize multithreading and Rust based UDFs. For complex python UDFs we could have a dedicated multiprocessing pipeline for just the transform. Or we could just utilize multithreading. In both cases we would exclude the I/O stage from the multiprocessing because that ends up being very memory hungry and inefficient. --------- Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>	2026-07-06 05:50:45 -07:00
dependabot[bot]	f84190fe12	chore(deps): bump the rust-minor-patch group with 2 updates (#3621 ) Bumps the rust-minor-patch group with 2 updates: [napi](https://github.com/napi-rs/napi-rs) and [napi-derive](https://github.com/napi-rs/napi-rs). Updates `napi` from 3.9.4 to 3.10.3 <details> <summary>Release notes</summary> <p><em>Sourced from <a href="https://github.com/napi-rs/napi-rs/releases">napi's releases</a>.</em></p> <blockquote> <h2>napi-v3.10.3</h2> <h3>Fixed</h3> <ul> <li><em>(napi)</em> preserve the JS error object when cloning an Error off-thread (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3375">#3375</a>)</li> </ul> <h2>napi-v3.10.2</h2> <h3>Fixed</h3> <ul> <li><em>(napi)</em> keep message and cause when cloning a JS-exception Error off-thread (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3373">#3373</a>)</li> </ul> <h2>napi-v3.10.1</h2> <h3>Fixed</h3> <ul> <li><em>(napi)</em> release Error's exception reference via the custom GC when dropped off-thread. (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3370">#3370</a>)</li> <li><em>(napi)</em> stop ref exception object in ThreadsafeFunction sync-throw path on wasm targets (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3369">#3369</a>)</li> </ul> <h3>Other</h3> <ul> <li><em>(napi)</em> share class accessor trampolines (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3364">#3364</a>)</li> <li>optimize object field raw property access (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3365">#3365</a>)</li> </ul> <h2>napi-v3.10.0</h2> <h3>Added</h3> <ul> <li><em>(napi)</em> implement <code>To</code>/<code>FromNapiValue</code> for <code>OsString</code>, <code>OsStr</code>, <code>Path</code> and <code>PathBuf</code> (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3339">#3339</a>)</li> </ul> <h3>Fixed</h3> <ul> <li><em>(napi)</em> route custom-GC Buffer/TypedArray cross-thread drops through the owning isolate (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3357">#3357</a>) (<a href="https://redirect.github.com/napi-rs/napi-rs/pull/3360">#3360</a>)</li> </ul> </blockquote> </details> <details> <summary>Commits</summary> <ul> <li><a href="https://github.com/napi-rs/napi-rs/commit/1ac467e06e71f78b983630926c7908894d08e496"><code>1ac467e</code></a> chore(napi): release v3.10.3 (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3376">#3376</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/9d672f9f9ac4784364548cac55c15444f4d2b1f8"><code>9d672f9</code></a> fix(napi): preserve the JS error object when cloning an Error off-thread (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3375">#3375</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/35476aebcc774a33b7e79e79d6c476db88a50215"><code>35476ae</code></a> chore(napi): release v3.10.2 (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3374">#3374</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/7844c7343f92f3ef45f2756dbba224c342a8467e"><code>7844c73</code></a> ci: dogfood script-jail <a href="https://github.com/v0"><code>@v0</code></a>.2.10 (lifecycle audit gate + safe install) (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3343">#3343</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/d449ccd8c50ad2268b051458ef919848e72b40a5"><code>d449ccd</code></a> fix(napi): keep message and cause when cloning a JS-exception Error off-threa...</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/2ec02a67a0ffdbe8dcbe93f7f24d1d79b861216b"><code>2ec02a6</code></a> chore(deps): update dependency electron to v43 (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3361">#3361</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/fd0a99f83015d4b67a591641d9ce66edf08d9740"><code>fd0a99f</code></a> chore(deps): update dependency <code>@types/sinon</code> to v22 (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3366">#3366</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/745cd8561f9be2781cc04c8ba4564c8f436792c1"><code>745cd85</code></a> fix: de-flake Windows CI (ava import-from-project EPERM race + cli e2e timeou...</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/2785de583a97e49adea8194090fca2ee12f067c8"><code>2785de5</code></a> chore: release (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3367">#3367</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/441ae7a7b6ddb06a2682a7dd27cf186a8afca9e8"><code>441ae7a</code></a> fix(napi): release Error's exception reference via the custom GC when dropped...</li> <li>Additional commits viewable in <a href="https://github.com/napi-rs/napi-rs/compare/napi-v3.9.4...napi-v3.10.3">compare view</a></li> </ul> </details> <br /> Updates `napi-derive` from 3.5.7 to 3.5.9 <details> <summary>Release notes</summary> <p><em>Sourced from <a href="https://github.com/napi-rs/napi-rs/releases">napi-derive's releases</a>.</em></p> <blockquote> <h2>napi-derive-v3.5.9</h2> <h3>Other</h3> <ul> <li>updated the following local packages: napi-derive-backend</li> </ul> <h2>napi-derive-v3.5.8</h2> <h3>Other</h3> <ul> <li>updated the following local packages: napi-derive-backend</li> </ul> </blockquote> </details> <details> <summary>Commits</summary> <ul> <li><a href="https://github.com/napi-rs/napi-rs/commit/2785de583a97e49adea8194090fca2ee12f067c8"><code>2785de5</code></a> chore: release (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3367">#3367</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/441ae7a7b6ddb06a2682a7dd27cf186a8afca9e8"><code>441ae7a</code></a> fix(napi): release Error's exception reference via the custom GC when dropped...</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/cfa3b77ed50dd3639278b219f5d0f630c596cfac"><code>cfa3b77</code></a> fix(deps): update emnapi to v1.11.2 (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3371">#3371</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/65918a6d195fa007985c83baf97a9ce82a95c2cf"><code>65918a6</code></a> fix(napi): stop ref exception object in ThreadsafeFunction sync-throw path on...</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/324c5502fb4deaabd6d76253e8a8e380c5a2bbb5"><code>324c550</code></a> perf(napi): share class accessor trampolines (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3364">#3364</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/80caf6063deb42468f2742bee02cc43ecb2e111d"><code>80caf60</code></a> perf: optimize object field raw property access (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3365">#3365</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/f72afd58976a83bb0776c6a71171673d94e82226"><code>f72afd5</code></a> chore: release (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3354">#3354</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/4effa4da6247a91048ca3462f2ff8eccdcfabfa4"><code>4effa4d</code></a> chore(deps): lock file maintenance (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3363">#3363</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/f2bf197f629e491362d1911578c57c33be2e561f"><code>f2bf197</code></a> chore(deps): lock file maintenance (<a href="https://redirect.github.com/napi-rs/napi-rs/issues/3362">#3362</a>)</li> <li><a href="https://github.com/napi-rs/napi-rs/commit/962a2f0504517c0f83ff7357100c8b5fc26203af"><code>962a2f0</code></a> fix(napi): route custom-GC Buffer/TypedArray cross-thread drops through the o...</li> <li>Additional commits viewable in <a href="https://github.com/napi-rs/napi-rs/compare/napi-derive-v3.5.7...napi-derive-v3.5.9">compare view</a></li> </ul> </details> <br /> Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`. [//]: # (dependabot-automerge-start) [//]: # (dependabot-automerge-end) --- <details> <summary>Dependabot commands and options</summary> <br /> You can trigger Dependabot actions by commenting on this PR: - `@dependabot rebase` will rebase this PR - `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it - `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency - `@dependabot ignore <dependency name> major version` will close this group update PR and stop Dependabot creating any more for the specific dependency's major version (unless you unignore this specific dependency's major version or upgrade to it yourself) - `@dependabot ignore <dependency name> minor version` will close this group update PR and stop Dependabot creating any more for the specific dependency's minor version (unless you unignore this specific dependency's minor version or upgrade to it yourself) - `@dependabot ignore <dependency name>` will close this group update PR and stop Dependabot creating any more for the specific dependency (unless you unignore this specific dependency or upgrade to it yourself) - `@dependabot unignore <dependency name>` will remove all of the ignore conditions of the specified dependency - `@dependabot unignore <dependency name> <ignore condition>` will remove the ignore condition of the specified dependency and ignore conditions </details> Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>	2026-07-04 11:11:10 -07:00
Weston Pace	122dcd0f66	chore: ignore RUSTSEC-2026-0194 and RUSTSEC-2026-0195 in cargo deny (#3616 ) quick-xml < 0.41.0 has two DoS advisories (quadratic attribute-name check and unbounded namespace allocation in NsReader). All three versions in our lockfile (0.26.0, 0.38.4, 0.39.4) are below the patched threshold. These are pulled in transitively by inferno (dev-only flame-graph dep), lance-namespace-impls (git dep from lance), and opendal/reqsign (cloud storage XML parsing). None of these paths expose attacker- controlled XML; clearing them requires upstream to upgrade to quick-xml >= 0.41.0. Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>	2026-07-02 16:54:34 -07:00
Eric B	e6661a7285	fix: handle empty/wrong-length vectors returned by embedding functions (#3192 ) ## Summary - When an embedding function returns an empty list (e.g. `[]`) for an input row — as can happen when a model produces no output for a blank string — `_append_vector_columns` crashed with `ArrowInvalid: Length of item not correct: expected N but got array of size 0` because PyArrow cannot fit a zero-length value into a fixed-size list element. - The fix adds a validation step in `gen()`, inside `_append_vector_columns`, that replaces any vector whose length does not match the expected `ndims` (including empty lists and `None`) with `None` before `pa.array()` is called. - `None` is a valid null in a PyArrow fixed-size list array, so the bad entry flows into `_handle_bad_vectors` and is handled according to the caller-supplied `on_bad_vectors` policy (`error` / `drop` / `fill` / `null`) instead of causing an unconditional crash. ## Test plan - [ ] Added `test_embedding_with_empty_output_vectors` in `python/python/tests/test_embeddings.py` that uses an embedding function returning `[]` for empty-string inputs, calls `table.add(..., on_bad_vectors="drop")`, and asserts no crash and that bad rows are correctly dropped. - [ ] Existing `test_embedding_with_bad_results` continues to pass (NaN vectors still handled correctly). - [ ] Verified manually that `pa.array([[1.,2.,3.,4.], []], type=pa.list_(pa.float32(), 4))` raises `ArrowInvalid` without the fix, and succeeds with `None` in place of `[]`. Fixes #1672 --------- Co-authored-by: Will Jones <willjones127@gmail.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>	2026-07-02 13:31:16 -07:00
Lance Release	37466a0390	Bump version: 0.31.0-beta.5 → 0.31.0-beta.6	2026-07-02 11:33:53 +00:00