From 678379561ed69265670536dab5ac6046f3f9b18f Mon Sep 17 00:00:00 2001
From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com>
Date: Mon, 6 Jul 2026 16:39:03 -0700
Subject: [PATCH 001/194] chore(deps): bump cmov from 0.5.3 to 0.5.4 in
/test_data/fri_straddle_pre_6610/datagen (#7594)
Bumps [cmov](https://github.com/RustCrypto/utils) from 0.5.3 to 0.5.4.
Commits
[](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)
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)
---
Dependabot commands and options
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 ignore conditions` will show all
of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop
Dependabot creating any more for this major version (unless you reopen
the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop
Dependabot creating any more for this minor version (unless you reopen
the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop
Dependabot creating any more for this dependency (unless you reopen the
PR or upgrade to it yourself)
You can disable automated security fix PRs for this repo from the
[Security Alerts
page](https://github.com/lance-format/lance/network/alerts).
Signed-off-by: dependabot[bot]
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
---
test_data/fri_straddle_pre_6610/datagen/Cargo.lock | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/test_data/fri_straddle_pre_6610/datagen/Cargo.lock b/test_data/fri_straddle_pre_6610/datagen/Cargo.lock
index 6fcff711a06..5edd16c2fed 100644
--- a/test_data/fri_straddle_pre_6610/datagen/Cargo.lock
+++ b/test_data/fri_straddle_pre_6610/datagen/Cargo.lock
@@ -1124,9 +1124,9 @@ dependencies = [
[[package]]
name = "cmov"
-version = "0.5.3"
+version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3f88a43d011fc4a6876cb7344703e297c71dda42494fee094d5f7c76bf13f746"
+checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
[[package]]
name = "colorchoice"
From e8d4e3fa019735c9a15dce504f4743bec8405786 Mon Sep 17 00:00:00 2001
From: Dan Rammer
Date: Mon, 6 Jul 2026 19:25:05 -0500
Subject: [PATCH 002/194] fix(mem_wal): use slice-aware size estimate for
memtable flush threshold (#7563)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
## Problem
`StoredBatch::estimate_batch_size` (mem_wal `batch_store.rs`) summed
`Array::get_array_memory_size`, which counts every buffer's full
`capacity()` **regardless of the array's offset/length**. Arrow's own
docs are explicit: a sliced `ArrayData` "may only refer to a subset of
the data ... but the size returned includes the entire size of the
buffers," and slices sharing a buffer "will both report the same size."
When ingest slices one incoming batch into N zero-copy WAL chunks, each
slice therefore reports the **whole shared parent buffer**. This
estimate feeds `maybe_trigger_memtable_flush` (`estimated_size() >=
max_memtable_size`), so the active memtable flushed far below the
configured size — e.g. a 250 MB target tripping at a few MB of real
data.
## Fix
Use Arrow's slice-aware `ArrayData::get_slice_memory_size`, which
reports only the slice's own window, with a fallback to the old call for
the rare types where it errors (conservative — over-counts, never
under).
## Evidence
A standalone repro (100 zero-copy slices of a 20 MB parent, the common
ingest pattern):
| | estimate |
|---|---|
| actual retained heap | ~20 MB |
| old (`get_array_memory_size`) | **~2744 MB** (131× over-count) |
| new (`get_slice_memory_size`) | ~20 MB |
Even a single owned (non-sliced) batch over-counts ~1.26× under the old
path, so the new estimate is strictly more accurate.
## Test
Adds `test_estimated_size_is_slice_aware`: tiles a parent into 100
zero-copy slices, appends them, and asserts the store's running estimate
covers the real payload without the ~N× blow-up (guarded against the old
over-counting sum).
`cargo test -p lance --lib batch_store`, `cargo fmt --all`, and `cargo
clippy -p lance --tests -- -D warnings` all pass.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 4.8 (1M context)
---
.../dataset/mem_wal/memtable/batch_store.rs | 162 +++++++++++++++++-
1 file changed, 161 insertions(+), 1 deletion(-)
diff --git a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
index 054d9b1630e..30e47e1a9bb 100644
--- a/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
+++ b/rust/lance/src/dataset/mem_wal/memtable/batch_store.rs
@@ -43,7 +43,9 @@ use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::sync::atomic::{AtomicUsize, Ordering};
+use arrow::array::ArrayData;
use arrow_array::RecordBatch;
+use arrow_schema::DataType;
/// A batch stored in the lock-free store.
#[derive(Clone)]
@@ -75,14 +77,56 @@ impl StoredBatch {
}
/// Estimate the memory size of a RecordBatch.
+ ///
+ /// Sums each column's slice-aware buffer size (see
+ /// [`Self::estimate_array_size`]) plus the struct overhead, so a column that
+ /// is a zero-copy slice of a larger parent contributes only its own window
+ /// rather than the whole shared buffer.
fn estimate_batch_size(batch: &RecordBatch) -> usize {
batch
.columns()
.iter()
- .map(|col| col.get_array_memory_size())
+ .map(|col| Self::estimate_array_size(&col.to_data()))
.sum::()
+ std::mem::size_of::()
}
+
+ /// Slice-aware buffer size of a single array.
+ ///
+ /// [`ArrayData::get_slice_memory_size`] reports each buffer's own window
+ /// (not the whole shared buffer), but omits the variadic data buffers of
+ /// `Utf8View`/`BinaryView` (values > 12 bytes) while still returning `Ok`, so
+ /// [`Self::view_data_buffers_size`] adds them. Those buffers are shared across
+ /// zero-copy slices and are counted at full capacity for each slice — an
+ /// over-count in the safe direction.
+ fn estimate_array_size(data: &ArrayData) -> usize {
+ match data.get_slice_memory_size() {
+ Ok(size) => size + Self::view_data_buffers_size(data),
+ // Fall back to the full-buffer sum for layouts the slice-aware call
+ // cannot handle.
+ Err(_) => data.get_array_memory_size(),
+ }
+ }
+
+ /// Capacity of the variadic `Utf8View`/`BinaryView` data buffers that
+ /// [`ArrayData::get_slice_memory_size`] omits, summed recursively over children.
+ fn view_data_buffers_size(data: &ArrayData) -> usize {
+ let mut size = 0;
+ if matches!(data.data_type(), DataType::Utf8View | DataType::BinaryView) {
+ // buffers()[0] is the 16-byte view array that get_slice_memory_size
+ // already counts; [1..] are the data buffers it skips.
+ size += data
+ .buffers()
+ .iter()
+ .skip(1)
+ .map(|b| b.capacity())
+ .sum::();
+ }
+ for child in data.child_data() {
+ size += Self::view_data_buffers_size(child);
+ }
+ size
+ }
}
/// Snapshot of the active batches that have not yet been flushed to WAL.
@@ -972,6 +1016,122 @@ mod tests {
assert_eq!(cap, 16); // minimum
}
+ #[test]
+ fn test_estimated_size_is_slice_aware() {
+ // A batch that is a zero-copy slice of a larger parent must contribute
+ // only its own window to the estimate, not the whole shared buffer.
+ // `get_array_memory_size` counts every buffer's full capacity regardless
+ // of offset/length, so N slices tiling one parent each report the
+ // parent's size and inflate the memtable estimate ~N×, tripping the
+ // flush threshold far below the configured size.
+ let chunk = 1_000;
+ let num_slices = 100;
+ let parent = create_test_batch(chunk * num_slices);
+
+ // One window vs an equivalently-sized owned batch should track each
+ // other; the buggy per-slice estimate would be ~num_slices× larger.
+ let slice_est = StoredBatch::estimate_batch_size(&parent.slice(0, chunk));
+ let owned_est = StoredBatch::estimate_batch_size(&create_test_batch(chunk));
+ assert!(
+ slice_est <= owned_est * 2,
+ "slice estimate {slice_est} should track its own window (~{owned_est}), not the parent"
+ );
+
+ // End-to-end: tiling the parent with zero-copy slices must not multiply
+ // the store's running estimate. Track what the old full-buffer behavior
+ // would have summed to for contrast.
+ let store = BatchStore::with_capacity(num_slices);
+ let mut over_counting_sum = 0usize;
+ for k in 0..num_slices {
+ let s = parent.slice(k * chunk, chunk);
+ over_counting_sum += s
+ .columns()
+ .iter()
+ .map(|col| col.get_array_memory_size())
+ .sum::()
+ + std::mem::size_of::();
+ store.append(s).unwrap();
+ }
+
+ // Two non-nullable Int32 columns → exactly 4 bytes/row/col of payload.
+ let payload_bytes = num_slices * chunk * 2 * std::mem::size_of::();
+ let estimated = store.estimated_bytes();
+ assert!(
+ estimated >= payload_bytes,
+ "estimate {estimated} should cover the actual payload {payload_bytes}"
+ );
+ // The old behavior over-counts by ~num_slices×; the fix must be far
+ // below it (generous 10× margin against struct/alignment overhead).
+ assert!(
+ estimated * 10 < over_counting_sum,
+ "estimate {estimated} should be far below the over-counting sum {over_counting_sum}"
+ );
+ }
+
+ #[test]
+ fn test_estimated_size_counts_view_data_buffers() {
+ // Long Utf8View/BinaryView values live in variadic data buffers that
+ // `get_slice_memory_size` ignores (returning ~16 * rows). The estimate
+ // must include them, both for a top-level view column and for a view
+ // array nested in a container, which is only reached via child_data
+ // recursion.
+ use arrow_array::{Array, ArrayRef, StringViewArray, StructArray};
+
+ let num_rows = 1_000;
+ // Each value exceeds the 12-byte inline limit, so it spills to a data buffer.
+ let long_value = "x".repeat(64);
+ let payload_bytes = num_rows * long_value.len();
+ // What the slice-aware call alone reports: just the 16-byte view entries.
+ let view_entries_only = num_rows * 16;
+
+ let make_views = || {
+ StringViewArray::from(
+ (0..num_rows)
+ .map(|_| Some(long_value.as_str()))
+ .collect::>(),
+ )
+ };
+ let assert_covers = |batch: &RecordBatch| {
+ let estimated = StoredBatch::estimate_batch_size(batch);
+ assert!(
+ estimated >= payload_bytes,
+ "estimate {estimated} should cover the view data-buffer payload {payload_bytes}"
+ );
+ assert!(
+ estimated > view_entries_only * 2,
+ "estimate {estimated} must exceed the ~{view_entries_only}-byte view-entry-only undercount"
+ );
+ };
+
+ // Top-level view column.
+ let flat = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![Field::new(
+ "s",
+ DataType::Utf8View,
+ false,
+ )])),
+ vec![Arc::new(make_views())],
+ )
+ .unwrap();
+ assert_covers(&flat);
+
+ // View nested inside a struct — reachable only through child_data recursion.
+ let nested = StructArray::from(vec![(
+ Arc::new(Field::new("s", DataType::Utf8View, false)),
+ Arc::new(make_views()) as ArrayRef,
+ )]);
+ let nested = RecordBatch::try_new(
+ Arc::new(ArrowSchema::new(vec![Field::new(
+ "st",
+ nested.data_type().clone(),
+ false,
+ )])),
+ vec![Arc::new(nested)],
+ )
+ .unwrap();
+ assert_covers(&nested);
+ }
+
#[test]
fn test_to_vec() {
let store = BatchStore::with_capacity(10);
From 9cbc09a580666beee3ac41212db8fce047fbfc14 Mon Sep 17 00:00:00 2001
From: Will Jones
Date: Mon, 6 Jul 2026 20:47:29 -0700
Subject: [PATCH 003/194] fix(deps): bump crossbeam-epoch to 0.9.20 for
RUSTSEC-2026-0204 (#7644)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
`crossbeam-epoch` 0.9.18 has an invalid pointer dereference in the
`fmt::Pointer` impl for `Atomic`/`Shared` when the underlying pointer is
invalid
([RUSTSEC-2026-0204](https://rustsec.org/advisories/RUSTSEC-2026-0204)).
It is pulled in transitively via `rayon` → `crossbeam-deque`, and is now
failing the `cargo-deny` CI check on all PRs.
This bumps `crossbeam-epoch` to the fixed version (>=0.9.20) across all
three lockfiles (`Cargo.lock`, `python/Cargo.lock`,
`java/lance-jni/Cargo.lock`).
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Claude Opus 4.8 (1M context)
---
Cargo.lock | 4 ++--
java/lance-jni/Cargo.lock | 4 ++--
python/Cargo.lock | 4 ++--
3 files changed, 6 insertions(+), 6 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index f537f382479..837a52b5d56 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1734,9 +1734,9 @@ dependencies = [
[[package]]
name = "crossbeam-epoch"
-version = "0.9.18"
+version = "0.9.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
dependencies = [
"crossbeam-utils",
]
diff --git a/java/lance-jni/Cargo.lock b/java/lance-jni/Cargo.lock
index e392fd82c76..c6e310a69ac 100644
--- a/java/lance-jni/Cargo.lock
+++ b/java/lance-jni/Cargo.lock
@@ -1385,9 +1385,9 @@ dependencies = [
[[package]]
name = "crossbeam-epoch"
-version = "0.9.18"
+version = "0.9.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
dependencies = [
"crossbeam-utils",
]
diff --git a/python/Cargo.lock b/python/Cargo.lock
index 440538da1ee..b609d4e8dbb 100644
--- a/python/Cargo.lock
+++ b/python/Cargo.lock
@@ -1583,9 +1583,9 @@ dependencies = [
[[package]]
name = "crossbeam-epoch"
-version = "0.9.18"
+version = "0.9.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e"
+checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f"
dependencies = [
"crossbeam-utils",
]
From 9c04698551db8ff7f64dc005ad2a4256f55b7441 Mon Sep 17 00:00:00 2001
From: Lance Release Bot
Date: Tue, 7 Jul 2026 03:49:18 +0000
Subject: [PATCH 004/194] chore: release beta version 9.0.0-beta.17
---
.bumpversion.toml | 2 +-
Cargo.lock | 48 +++++++++++++++++++--------------------
Cargo.toml | 44 +++++++++++++++++------------------
java/lance-jni/Cargo.lock | 40 ++++++++++++++++----------------
java/lance-jni/Cargo.toml | 2 +-
java/pom.xml | 2 +-
python/Cargo.lock | 40 ++++++++++++++++----------------
python/Cargo.toml | 2 +-
8 files changed, 90 insertions(+), 90 deletions(-)
diff --git a/.bumpversion.toml b/.bumpversion.toml
index 6840e4f3138..ce1aad8e3e7 100644
--- a/.bumpversion.toml
+++ b/.bumpversion.toml
@@ -1,5 +1,5 @@
[tool.bumpversion]
-current_version = "9.0.0-beta.16"
+current_version = "9.0.0-beta.17"
parse = "(?P\\d+)\\.(?P\\d+)\\.(?P\\d+)(-(?P(beta|rc))\\.(?P\\d+))?"
serialize = [
"{major}.{minor}.{patch}-{prerelease}.{prerelease_num}",
diff --git a/Cargo.lock b/Cargo.lock
index 837a52b5d56..7ba3cd00312 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3076,7 +3076,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"rand 0.9.4",
@@ -4380,7 +4380,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"all_asserts",
"approx",
@@ -4483,7 +4483,7 @@ dependencies = [
[[package]]
name = "lance-arrow"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4531,7 +4531,7 @@ dependencies = [
[[package]]
name = "lance-bitpacking"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrayref",
"bitpacking",
@@ -4542,7 +4542,7 @@ dependencies = [
[[package]]
name = "lance-core"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4582,7 +4582,7 @@ dependencies = [
[[package]]
name = "lance-datafusion"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4615,7 +4615,7 @@ dependencies = [
[[package]]
name = "lance-datagen"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4634,7 +4634,7 @@ dependencies = [
[[package]]
name = "lance-derive"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"proc-macro2",
"quote",
@@ -4643,7 +4643,7 @@ dependencies = [
[[package]]
name = "lance-encoding"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -4688,7 +4688,7 @@ dependencies = [
[[package]]
name = "lance-examples"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"all_asserts",
"arrow",
@@ -4714,7 +4714,7 @@ dependencies = [
[[package]]
name = "lance-file"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -4753,7 +4753,7 @@ dependencies = [
[[package]]
name = "lance-geo"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"datafusion",
"geo-traits",
@@ -4767,7 +4767,7 @@ dependencies = [
[[package]]
name = "lance-index"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"approx",
"arc-swap",
@@ -4844,7 +4844,7 @@ dependencies = [
[[package]]
name = "lance-io"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-arith",
@@ -4892,7 +4892,7 @@ dependencies = [
[[package]]
name = "lance-linalg"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"approx",
"arrow-array",
@@ -4912,7 +4912,7 @@ dependencies = [
[[package]]
name = "lance-namespace"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"async-trait",
@@ -4924,7 +4924,7 @@ dependencies = [
[[package]]
name = "lance-namespace-datafusion"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -4940,7 +4940,7 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -5004,7 +5004,7 @@ dependencies = [
[[package]]
name = "lance-select"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -5022,7 +5022,7 @@ dependencies = [
[[package]]
name = "lance-table"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -5068,7 +5068,7 @@ dependencies = [
[[package]]
name = "lance-test-macros"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"proc-macro2",
"quote",
@@ -5077,7 +5077,7 @@ dependencies = [
[[package]]
name = "lance-testing"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-schema",
@@ -5090,7 +5090,7 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"icu_segmenter",
"jieba-rs",
@@ -5103,7 +5103,7 @@ dependencies = [
[[package]]
name = "lance-tools"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"clap",
"lance-core",
diff --git a/Cargo.toml b/Cargo.toml
index 486aa1e56a2..7755010da19 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -32,7 +32,7 @@ resolver = "3"
[workspace.package]
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
edition = "2024"
authors = ["Lance Devs "]
license = "Apache-2.0"
@@ -57,27 +57,27 @@ rust-version = "1.91.0"
[workspace.dependencies]
arc-swap = "1.7"
libc = "0.2.176"
-lance = { version = "=9.0.0-beta.16", path = "./rust/lance", default-features = false }
-lance-arrow = { version = "=9.0.0-beta.16", path = "./rust/lance-arrow" }
-lance-core = { version = "=9.0.0-beta.16", path = "./rust/lance-core" }
-lance-datafusion = { version = "=9.0.0-beta.16", path = "./rust/lance-datafusion" }
-lance-datagen = { version = "=9.0.0-beta.16", path = "./rust/lance-datagen" }
-lance-derive = { version = "=9.0.0-beta.16", path = "./rust/lance-derive" }
-lance-encoding = { version = "=9.0.0-beta.16", path = "./rust/lance-encoding" }
-lance-file = { version = "=9.0.0-beta.16", path = "./rust/lance-file" }
-lance-geo = { version = "=9.0.0-beta.16", path = "./rust/lance-geo" }
-lance-index = { version = "=9.0.0-beta.16", path = "./rust/lance-index" }
-lance-io = { version = "=9.0.0-beta.16", path = "./rust/lance-io", default-features = false }
-lance-linalg = { version = "=9.0.0-beta.16", path = "./rust/lance-linalg" }
-lance-namespace = { version = "=9.0.0-beta.16", path = "./rust/lance-namespace" }
-lance-namespace-impls = { version = "=9.0.0-beta.16", path = "./rust/lance-namespace-impls" }
+lance = { version = "=9.0.0-beta.17", path = "./rust/lance", default-features = false }
+lance-arrow = { version = "=9.0.0-beta.17", path = "./rust/lance-arrow" }
+lance-core = { version = "=9.0.0-beta.17", path = "./rust/lance-core" }
+lance-datafusion = { version = "=9.0.0-beta.17", path = "./rust/lance-datafusion" }
+lance-datagen = { version = "=9.0.0-beta.17", path = "./rust/lance-datagen" }
+lance-derive = { version = "=9.0.0-beta.17", path = "./rust/lance-derive" }
+lance-encoding = { version = "=9.0.0-beta.17", path = "./rust/lance-encoding" }
+lance-file = { version = "=9.0.0-beta.17", path = "./rust/lance-file" }
+lance-geo = { version = "=9.0.0-beta.17", path = "./rust/lance-geo" }
+lance-index = { version = "=9.0.0-beta.17", path = "./rust/lance-index" }
+lance-io = { version = "=9.0.0-beta.17", path = "./rust/lance-io", default-features = false }
+lance-linalg = { version = "=9.0.0-beta.17", path = "./rust/lance-linalg" }
+lance-namespace = { version = "=9.0.0-beta.17", path = "./rust/lance-namespace" }
+lance-namespace-impls = { version = "=9.0.0-beta.17", path = "./rust/lance-namespace-impls" }
lance-namespace-datafusion = { version = "=7.0.0-beta.9", path = "./rust/lance-namespace-datafusion" }
lance-namespace-reqwest-client = "0.8.6"
-lance-select = { version = "=9.0.0-beta.16", path = "./rust/lance-select" }
-lance-tokenizer = { version = "=9.0.0-beta.16", path = "./rust/lance-tokenizer" }
-lance-table = { version = "=9.0.0-beta.16", path = "./rust/lance-table" }
-lance-test-macros = { version = "=9.0.0-beta.16", path = "./rust/lance-test-macros" }
-lance-testing = { version = "=9.0.0-beta.16", path = "./rust/lance-testing" }
+lance-select = { version = "=9.0.0-beta.17", path = "./rust/lance-select" }
+lance-tokenizer = { version = "=9.0.0-beta.17", path = "./rust/lance-tokenizer" }
+lance-table = { version = "=9.0.0-beta.17", path = "./rust/lance-table" }
+lance-test-macros = { version = "=9.0.0-beta.17", path = "./rust/lance-test-macros" }
+lance-testing = { version = "=9.0.0-beta.17", path = "./rust/lance-testing" }
approx = "0.5.1"
# Note that this one does not include pyarrow
arrow = { version = "58.0.0", optional = false, features = ["prettyprint"] }
@@ -104,7 +104,7 @@ half = { "version" = "2.1", default-features = false, features = [
"num-traits",
"std",
] }
-lance-bitpacking = { version = "=9.0.0-beta.16", path = "./rust/compression/bitpacking" }
+lance-bitpacking = { version = "=9.0.0-beta.17", path = "./rust/compression/bitpacking" }
bitpacking = "0.9"
bitvec = "1"
bytes = "1.11.1"
@@ -143,7 +143,7 @@ datafusion-substrait = { version = "53.0.0", default-features = false }
dirs = "6.0.0"
either = "1.0"
fst = { version = "0.4.7", features = ["levenshtein"] }
-fsst = { version = "=9.0.0-beta.16", path = "./rust/compression/fsst" }
+fsst = { version = "=9.0.0-beta.17", path = "./rust/compression/fsst" }
futures = "0.3"
geoarrow-array = "0.8"
geoarrow-schema = "0.8"
diff --git a/java/lance-jni/Cargo.lock b/java/lance-jni/Cargo.lock
index c6e310a69ac..43e7b8b1b9a 100644
--- a/java/lance-jni/Cargo.lock
+++ b/java/lance-jni/Cargo.lock
@@ -2470,7 +2470,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"rand 0.9.4",
@@ -3647,7 +3647,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arc-swap",
"arrow",
@@ -3720,7 +3720,7 @@ dependencies = [
[[package]]
name = "lance-arrow"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -3762,7 +3762,7 @@ dependencies = [
[[package]]
name = "lance-bitpacking"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrayref",
"crunchy",
@@ -3772,7 +3772,7 @@ dependencies = [
[[package]]
name = "lance-core"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -3810,7 +3810,7 @@ dependencies = [
[[package]]
name = "lance-datafusion"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -3842,7 +3842,7 @@ dependencies = [
[[package]]
name = "lance-datagen"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -3859,7 +3859,7 @@ dependencies = [
[[package]]
name = "lance-derive"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"proc-macro2",
"quote",
@@ -3868,7 +3868,7 @@ dependencies = [
[[package]]
name = "lance-encoding"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -3903,7 +3903,7 @@ dependencies = [
[[package]]
name = "lance-file"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -3933,7 +3933,7 @@ dependencies = [
[[package]]
name = "lance-geo"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"datafusion",
"geo-traits",
@@ -3947,7 +3947,7 @@ dependencies = [
[[package]]
name = "lance-index"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arc-swap",
"arrow",
@@ -4015,7 +4015,7 @@ dependencies = [
[[package]]
name = "lance-io"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-arith",
@@ -4056,7 +4056,7 @@ dependencies = [
[[package]]
name = "lance-jni"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4092,7 +4092,7 @@ dependencies = [
[[package]]
name = "lance-linalg"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4108,7 +4108,7 @@ dependencies = [
[[package]]
name = "lance-namespace"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"async-trait",
@@ -4120,7 +4120,7 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-ipc",
@@ -4169,7 +4169,7 @@ dependencies = [
[[package]]
name = "lance-select"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4184,7 +4184,7 @@ dependencies = [
[[package]]
name = "lance-table"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4221,7 +4221,7 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"icu_segmenter",
"rust-stemmers",
diff --git a/java/lance-jni/Cargo.toml b/java/lance-jni/Cargo.toml
index d8f839b273a..c3d765d85af 100644
--- a/java/lance-jni/Cargo.toml
+++ b/java/lance-jni/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "lance-jni"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
edition = "2024"
authors = ["Lance Devs "]
rust-version = "1.91"
diff --git a/java/pom.xml b/java/pom.xml
index 89b567c5f3e..9565e595040 100644
--- a/java/pom.xml
+++ b/java/pom.xml
@@ -7,7 +7,7 @@
org.lance
lance-core
Lance Core
- 9.0.0-beta.16
+ 9.0.0-beta.17
jar
Lance Format Java API
diff --git a/python/Cargo.lock b/python/Cargo.lock
index b609d4e8dbb..1c371bf90cd 100644
--- a/python/Cargo.lock
+++ b/python/Cargo.lock
@@ -2850,7 +2850,7 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "fsst"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"rand 0.9.4",
@@ -4049,7 +4049,7 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a"
[[package]]
name = "lance"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arc-swap",
"arrow",
@@ -4123,7 +4123,7 @@ dependencies = [
[[package]]
name = "lance-arrow"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4165,7 +4165,7 @@ dependencies = [
[[package]]
name = "lance-bitpacking"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrayref",
"crunchy",
@@ -4175,7 +4175,7 @@ dependencies = [
[[package]]
name = "lance-core"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4213,7 +4213,7 @@ dependencies = [
[[package]]
name = "lance-datafusion"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4245,7 +4245,7 @@ dependencies = [
[[package]]
name = "lance-datagen"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4262,7 +4262,7 @@ dependencies = [
[[package]]
name = "lance-derive"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"proc-macro2",
"quote",
@@ -4271,7 +4271,7 @@ dependencies = [
[[package]]
name = "lance-encoding"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -4306,7 +4306,7 @@ dependencies = [
[[package]]
name = "lance-file"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-arith",
"arrow-array",
@@ -4336,7 +4336,7 @@ dependencies = [
[[package]]
name = "lance-geo"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"datafusion",
"geo-traits",
@@ -4350,7 +4350,7 @@ dependencies = [
[[package]]
name = "lance-index"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arc-swap",
"arrow",
@@ -4419,7 +4419,7 @@ dependencies = [
[[package]]
name = "lance-io"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-arith",
@@ -4460,7 +4460,7 @@ dependencies = [
[[package]]
name = "lance-linalg"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4476,7 +4476,7 @@ dependencies = [
[[package]]
name = "lance-namespace"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"async-trait",
@@ -4488,7 +4488,7 @@ dependencies = [
[[package]]
name = "lance-namespace-impls"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-ipc",
@@ -4537,7 +4537,7 @@ dependencies = [
[[package]]
name = "lance-select"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow-array",
"arrow-buffer",
@@ -4552,7 +4552,7 @@ dependencies = [
[[package]]
name = "lance-table"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"arrow",
"arrow-array",
@@ -4591,7 +4591,7 @@ dependencies = [
[[package]]
name = "lance-tokenizer"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"icu_segmenter",
"jieba-rs",
@@ -6029,7 +6029,7 @@ dependencies = [
[[package]]
name = "pylance"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
dependencies = [
"alloc-stdlib",
"arrow",
diff --git a/python/Cargo.toml b/python/Cargo.toml
index f0c84c01a16..9186496660d 100644
--- a/python/Cargo.toml
+++ b/python/Cargo.toml
@@ -1,6 +1,6 @@
[package]
name = "pylance"
-version = "9.0.0-beta.16"
+version = "9.0.0-beta.17"
edition = "2024"
authors = ["Lance Devs "]
license = "Apache-2.0"
From 6c291ab14555d327dda61860433034bce7586bd7 Mon Sep 17 00:00:00 2001
From: xloya <982052490@qq.com>
Date: Tue, 7 Jul 2026 14:00:22 +0800
Subject: [PATCH 005/194] fix: convert Arrow JSON to Lance JSON in
single-fragment create path (#7469)
## Problem
Creating a single fragment via `LanceFragment.create` /
`FragmentCreateBuilder` with an Arrow JSON column (`arrow.json`, stored
as Utf8) writes raw UTF-8 bytes into a column whose schema declares
Lance JSON (JSONB / LargeBinary), corrupting subsequent reads.
## Root cause
The multi-fragment and dataset write paths run the Arrow JSON -> Lance
JSON conversion via `do_write_fragments`. The single-fragment create
path skipped it.
## Fix
Run the same conversion in the create path via
`SchemaAdapter::to_physical_stream`.
## Test
`test_fragment_create_with_json_column` (Python).
Co-authored-by: xiaojiebao
---
python/python/tests/test_fragment.py | 34 ++++++++++++++++++++++++
rust/lance/src/dataset/fragment/write.rs | 7 +++++
2 files changed, 41 insertions(+)
diff --git a/python/python/tests/test_fragment.py b/python/python/tests/test_fragment.py
index b05888df31f..8b41e051b3d 100644
--- a/python/python/tests/test_fragment.py
+++ b/python/python/tests/test_fragment.py
@@ -865,3 +865,37 @@ def test_fragment_take_with_json_column(tmp_path):
assert metas[0] == '{"val":1}'
assert metas[1] == '{"val":4}'
assert metas[2] == '{"val":7}'
+
+
+def test_fragment_create_with_json_column(tmp_path):
+ """Test that LanceFragment.create works with Arrow JSON extension type.
+
+ Previously the single-fragment create path skipped the Arrow JSON (Utf8) ->
+ Lance JSON (JSONB LargeBinary) conversion that write_dataset/write_fragments
+ perform, so the raw UTF-8 string bytes were written into a column whose schema
+ declared JSONB. Reads then miss-decoded the bytes and returned garbage.
+ """
+ json_type = pa.json_()
+ data = pa.table(
+ {
+ "uid": pa.array(["a", "b", "c", "d"], type=pa.utf8()),
+ "payload": pa.array(
+ ['{"x":1}', '{"x":2}', '{"y":3}', '{"y":4}'],
+ type=json_type,
+ ),
+ }
+ )
+
+ frag = LanceFragment.create(tmp_path, data)
+ operation = LanceOperation.Overwrite(data.schema, [frag])
+ dataset = LanceDataset.commit(tmp_path, operation)
+
+ result = dataset.to_table()
+ assert result.column("uid").to_pylist() == ["a", "b", "c", "d"]
+ payloads = result.column("payload").to_pylist()
+ assert [json.loads(p) for p in payloads] == [
+ {"x": 1},
+ {"x": 2},
+ {"y": 3},
+ {"y": 4},
+ ]
diff --git a/rust/lance/src/dataset/fragment/write.rs b/rust/lance/src/dataset/fragment/write.rs
index a61f0e0c46a..1853ecceeac 100644
--- a/rust/lance/src/dataset/fragment/write.rs
+++ b/rust/lance/src/dataset/fragment/write.rs
@@ -21,6 +21,7 @@ use uuid::Uuid;
use crate::Result;
use crate::dataset::builder::DatasetBuilder;
+use crate::dataset::utils::SchemaAdapter;
use crate::dataset::write::{do_write_fragments, validate_and_resolve_target_bases_with_primary};
use crate::dataset::{DATA_DIR, Dataset, ReadParams, WriteMode, WriteParams};
@@ -106,6 +107,12 @@ impl<'a> FragmentCreateBuilder<'a> {
id: Option,
) -> Result {
let (stream, schema) = self.get_stream_and_schema(Box::new(source)).await?;
+ // Convert Arrow JSON columns (`arrow.json`, stored as Utf8) into Lance JSON
+ // (`lance.json`, stored as JSONB-encoded LargeBinary) before writing. The
+ // multi-fragment and dataset write paths perform this through `do_write_fragments`;
+ // the single-fragment create path must do the same or the raw UTF-8 string bytes
+ // would be written into a column whose schema declares JSONB, corrupting reads.
+ let stream = SchemaAdapter::new(stream.schema()).to_physical_stream(stream);
self.write_impl(stream, schema, id).await
}
From 9e7776ec6c9353cdf16649b6b59b23f8c5d89e72 Mon Sep 17 00:00:00 2001
From: Yang Cen
Date: Tue, 7 Jul 2026 17:47:49 +0800
Subject: [PATCH 006/194] fix(fts): enforce fuzzy max_expansions globally
across index partitions (#7634)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
Follow-up to the review discussion on #7601: fuzzy expansion previously
ran
inside every partition with a per-partition `max_expansions` cap, and
the
per-partition picks were unioned afterwards. The effective cap was
`num_partitions × max_expansions`, so the same query could match more
terms —
and return different results — purely because of how the corpus happened
to be
partitioned (e.g. after a tail-heavy build splits its leftovers).
## What
- **Expansion now runs once per query**, in
`InvertedIndex::bm25_search`,
under the query-wide `max_expansions` budget. For each query token the
per-partition FST candidates merge into one lexicographically ordered
set
and the remaining budget takes a prefix of it. Each partition is only
asked
for its `remaining` lexicographically smallest candidates, which is
lossless
for that selection: any term among the merged lex-smallest `remaining`
is
also among its own partition's smallest `remaining`.
- **Partitions receive the final token list** and no longer expand
(`load_posting_lists` drops its `expand_fuzzy` call); `params.fuzziness`
still drives the grouped fuzzy dedup/scoring semantics.
- **The BM25 scorer reuses the same expansion**
(`bm25_scorer_for_final_tokens`),
so scorer terms and searched terms stay in lockstep and a query pays for
one
expansion instead of one for the scorer plus one per partition.
- **Fuzzy AND/phrase keep their contract** — every original token
position
must retain at least one expansion — via an index-level check
(previously
implicit in each partition's own expansion).
- `InvertedPartition::expand_fuzzy` stays for API compatibility,
reimplemented on the shared per-token candidate collector;
single-partition
behavior is unchanged (same FST lexicographic order, same budget fill).
## Semantics
Single-partition indexes behave exactly as before. Multi-partition
indexes
previously over-expanded in proportion to their partition count; they
now
honor the documented cap, and the expansion (hence the result set) is a
pure
function of the segment's vocabulary, independent of partition shape.
## Tests
- `test_fuzzy_expansion_cap_is_global_across_partitions`: two partitions
with
disjoint variants, binding cap → exactly the 3 lexicographically
smallest
terms across both (fails on main, which returns 4).
- `test_fuzzy_results_independent_of_partition_shape`: the same four
docs
built as one partition and as two return identical `(row_id, score)`
sets
under a binding cap (fails on main).
- Existing fuzzy suite (grouped AND, position grouping, whole-query cap)
unchanged and passing. The non-binding-cap regression test over the real
tail-split builder path lives in #7601.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Yang Cen
---
rust/lance-index/src/scalar/inverted/index.rs | 328 ++++++++++++++----
1 file changed, 265 insertions(+), 63 deletions(-)
diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs
index d5a513c7112..bc9e7c9a871 100644
--- a/rust/lance-index/src/scalar/inverted/index.rs
+++ b/rust/lance-index/src/scalar/inverted/index.rs
@@ -10,7 +10,7 @@ use std::{
collections::BinaryHeap,
};
use std::{
- collections::{BTreeMap, HashMap, HashSet},
+ collections::{BTreeMap, BTreeSet, HashMap, HashSet},
ops::Range,
time::Instant,
};
@@ -630,22 +630,25 @@ impl InvertedIndex {
query_tokens: &Tokens,
params: &FtsSearchParams,
) -> Result {
- let (total_tokens, num_docs) = self.aggregate_corpus_stats().await?;
- let mut terms: Vec = Vec::new();
- let mut seen = HashSet::new();
if matches!(params.fuzziness, Some(n) if n != 0) {
let expanded = self.expand_fuzzy_tokens(query_tokens, params)?;
- for idx in 0..expanded.len() {
- let token = expanded.get_token(idx);
- if seen.insert(token.to_string()) {
- terms.push(token.to_string());
- }
- }
+ self.bm25_scorer_for_final_tokens(&expanded).await
} else {
- for token in query_tokens {
- if seen.insert(token.to_string()) {
- terms.push(token.to_string());
- }
+ self.bm25_scorer_for_final_tokens(query_tokens).await
+ }
+ }
+
+ /// Scorer for a token list that needs no further fuzzy expansion: dedup
+ /// the terms and pull their document frequencies. `bm25_search` calls
+ /// this with the tokens it already expanded, so the expansion runs once
+ /// per query rather than once for the scorer and once per partition.
+ async fn bm25_scorer_for_final_tokens(&self, tokens: &Tokens) -> Result {
+ let (total_tokens, num_docs) = self.aggregate_corpus_stats().await?;
+ let mut terms: Vec = Vec::new();
+ let mut seen = HashSet::new();
+ for token in tokens {
+ if seen.insert(token.to_string()) {
+ terms.push(token.to_string());
}
}
let mut token_docs = HashMap::with_capacity(terms.len());
@@ -717,17 +720,46 @@ impl InvertedIndex {
}
/// Expand fuzzy query tokens against all partitions in this segment.
+ ///
+ /// `params.max_expansions` caps the whole query's expansion, not any
+ /// single partition's: for each query token the per-partition candidates
+ /// (each streamed in FST key order) merge into one lexicographically
+ /// ordered set, and the remaining budget takes a prefix of it. The
+ /// selected terms are a pure function of the segment's vocabulary, so
+ /// splitting the same corpus into more partitions cannot change which
+ /// terms a fuzzy query matches.
pub fn expand_fuzzy_tokens(&self, tokens: &Tokens, params: &FtsSearchParams) -> Result {
let mut expanded_tokens = Vec::new();
let mut expanded_positions = Vec::new();
let mut seen = HashSet::new();
- for partition in &self.partitions {
- let expanded = partition.expand_fuzzy(tokens, params)?;
- for idx in 0..expanded.len() {
- let token = expanded.get_token(idx);
- let position = expanded.position(idx);
- if seen.insert((token.to_string(), position)) {
- expanded_tokens.push(token.to_string());
+ for token_idx in 0..tokens.len() {
+ let remaining = params.max_expansions.saturating_sub(expanded_tokens.len());
+ if remaining == 0 {
+ break;
+ }
+ let token = tokens.get_token(token_idx);
+ let position = tokens.position(token_idx);
+ // Each partition contributes at most its `remaining`
+ // lexicographically smallest candidates, so the global
+ // lex-smallest `remaining` selection below is unaffected by the
+ // per-partition truncation.
+ let mut candidates = BTreeSet::new();
+ let base_prefix_len = tokens.token_type().prefix_len(token) as u32;
+ for partition in &self.partitions {
+ partition.collect_fuzzy_candidates(
+ token,
+ base_prefix_len,
+ params,
+ remaining,
+ &mut candidates,
+ )?;
+ }
+ for candidate in candidates {
+ if expanded_tokens.len() >= params.max_expansions {
+ break;
+ }
+ if seen.insert((candidate.clone(), position)) {
+ expanded_tokens.push(candidate);
expanded_positions.push(position);
}
}
@@ -753,6 +785,28 @@ impl InvertedIndex {
metrics: Arc,
base_scorer: Option<&MemBM25Scorer>,
) -> Result<(Vec, Vec)> {
+ // Fuzzy expansion runs once here, with the global `max_expansions`
+ // budget, instead of once per partition: partitions receive the
+ // final token list, so the matched terms cannot depend on how the
+ // corpus happens to be partitioned.
+ let tokens = if matches!(params.fuzziness, Some(n) if n != 0) {
+ let expanded = Arc::new(self.expand_fuzzy_tokens(tokens.as_ref(), params.as_ref())?);
+ if operator == Operator::And || params.phrase_slop.is_some() {
+ // AND/phrase semantics require every original token position
+ // to keep at least one expansion; a position that expands to
+ // nothing anywhere in the segment can never be matched.
+ let surviving = (0..expanded.len())
+ .map(|idx| expanded.position(idx))
+ .collect::>();
+ if (0..tokens.len()).any(|idx| !surviving.contains(&tokens.position(idx))) {
+ return Ok((Vec::new(), Vec::new()));
+ }
+ }
+ expanded
+ } else {
+ tokens
+ };
+
// The wand only consults `scorer.doc_weight`, which is metadata-free.
// The outer aggregation below consults `scorer.query_weight`, which
// hits per-token `posting_len`; building a `MemBM25Scorer` with
@@ -761,9 +815,7 @@ impl InvertedIndex {
let scorer: &dyn Scorer = if let Some(base_scorer) = base_scorer {
base_scorer
} else {
- local_scorer = self
- .bm25_base_scorer(tokens.as_ref(), params.as_ref())
- .await?;
+ local_scorer = self.bm25_scorer_for_final_tokens(tokens.as_ref()).await?;
&local_scorer
};
@@ -1484,47 +1536,29 @@ impl InvertedPartition {
let mut new_positions = Vec::with_capacity(new_tokens.capacity());
let mut seen = HashSet::new();
for token_idx in 0..tokens.len() {
- if new_tokens.len() >= params.max_expansions {
+ let remaining = params.max_expansions.saturating_sub(new_tokens.len());
+ if remaining == 0 {
break;
}
let token = tokens.get_token(token_idx);
let position = tokens.position(token_idx);
- let fuzziness = match params.fuzziness {
- Some(fuzziness) => fuzziness,
- None => MatchQuery::auto_fuzziness(token),
- };
- let lev = fst::automaton::Levenshtein::new(token, fuzziness)
- .map_err(|e| Error::index(format!("failed to construct the fuzzy query: {}", e)))?;
-
- let base_len = tokens.token_type().prefix_len(token) as u32;
- if let TokenMap::Fst(ref map) = self.tokens.tokens {
- let mut expanded = Vec::new();
- let remaining = params.max_expansions - new_tokens.len();
- match base_len + params.prefix_length {
- 0 => take_fst_keys(map.search(lev), &mut expanded, remaining),
- prefix_length => {
- let prefix = &token[..min(prefix_length as usize, token.len())];
- let prefix = fst::automaton::Str::new(prefix).starts_with();
- take_fst_keys(
- map.search(lev.intersection(prefix)),
- &mut expanded,
- remaining,
- )
- }
+ let base_prefix_len = tokens.token_type().prefix_len(token) as u32;
+ let mut candidates = BTreeSet::new();
+ self.collect_fuzzy_candidates(
+ token,
+ base_prefix_len,
+ params,
+ remaining,
+ &mut candidates,
+ )?;
+ for candidate in candidates {
+ if new_tokens.len() >= params.max_expansions {
+ break;
}
- for token in expanded {
- if seen.insert((token.clone(), position)) {
- new_tokens.push(token);
- new_positions.push(position);
- if new_tokens.len() >= params.max_expansions {
- break;
- }
- }
+ if seen.insert((candidate.clone(), position)) {
+ new_tokens.push(candidate);
+ new_positions.push(position);
}
- } else {
- return Err(Error::index(
- "tokens is not fst, which is not expected".to_owned(),
- ));
}
}
Ok(Tokens::with_positions(
@@ -1534,6 +1568,47 @@ impl InvertedPartition {
))
}
+ /// Collect up to `limit` fuzzy candidates for one query token from this
+ /// partition's token FST, in key (lexicographic) order. Callers merge
+ /// candidates across partitions and apply the query-wide
+ /// `max_expansions` budget; truncating each partition at `limit` is
+ /// lossless for that selection because any term among the merged
+ /// lexicographically-smallest `limit` is also among its own partition's
+ /// smallest `limit`.
+ fn collect_fuzzy_candidates(
+ &self,
+ token: &str,
+ base_prefix_len: u32,
+ params: &FtsSearchParams,
+ limit: usize,
+ candidates: &mut BTreeSet,
+ ) -> Result<()> {
+ let fuzziness = match params.fuzziness {
+ Some(fuzziness) => fuzziness,
+ None => MatchQuery::auto_fuzziness(token),
+ };
+ let lev = fst::automaton::Levenshtein::new(token, fuzziness)
+ .map_err(|e| Error::index(format!("failed to construct the fuzzy query: {}", e)))?;
+
+ if let TokenMap::Fst(ref map) = self.tokens.tokens {
+ let mut expanded = Vec::new();
+ match base_prefix_len + params.prefix_length {
+ 0 => take_fst_keys(map.search(lev), &mut expanded, limit),
+ prefix_length => {
+ let prefix = &token[..min(prefix_length as usize, token.len())];
+ let prefix = fst::automaton::Str::new(prefix).starts_with();
+ take_fst_keys(map.search(lev.intersection(prefix)), &mut expanded, limit)
+ }
+ }
+ candidates.extend(expanded);
+ Ok(())
+ } else {
+ Err(Error::index(
+ "tokens is not fst, which is not expected".to_owned(),
+ ))
+ }
+ }
+
fn union_plain_posting_lists(postings: Vec) -> Result {
let mut freqs_by_row_id = BTreeMap::new();
for posting in postings {
@@ -1642,10 +1717,11 @@ impl InvertedPartition {
.map(|index| tokens.position(index))
.collect::>()
});
- let tokens = match is_fuzzy {
- true => self.expand_fuzzy(tokens, params)?,
- false => tokens.clone(),
- };
+ // Fuzzy expansion already ran once at the index level (see
+ // `InvertedIndex::bm25_search`) under the global `max_expansions`
+ // budget; the incoming tokens are final and `is_fuzzy` only drives
+ // the grouped dedup/scoring semantics below.
+ let tokens = tokens.clone();
let token_positions = (0..tokens.len())
.map(|index| tokens.position(index))
.collect::>();
@@ -8116,6 +8192,132 @@ mod tests {
);
}
+ /// Write one partition holding `variants` in order, with one
+ /// single-token doc per variant taken from `row_ids`.
+ async fn write_variant_partition(
+ store: &Arc,
+ partition_id: u64,
+ variants: &[&str],
+ row_ids: &[u64],
+ ) {
+ let mut builder = InnerBuilder::new(partition_id, false, TokenSetFormat::default());
+ for token in variants {
+ builder.tokens.add((*token).to_owned());
+ builder.posting_lists.push(PostingListBuilder::new(false));
+ }
+ for (local_idx, row_id) in row_ids.iter().enumerate() {
+ builder.posting_lists[local_idx].add(local_idx as u32, PositionRecorder::Count(1));
+ builder.docs.append(*row_id, 1);
+ }
+ builder.write(store.as_ref()).await.unwrap();
+ }
+
+ #[tokio::test]
+ async fn test_fuzzy_expansion_cap_is_global_across_partitions() {
+ let tmpdir = TempObjDir::default();
+ let store = Arc::new(LanceIndexStore::new(
+ ObjectStore::local().into(),
+ tmpdir.clone(),
+ Arc::new(LanceCache::no_cache()),
+ ));
+
+ write_variant_partition(&store, 0, &["alpha", "alphb"], &[100, 101]).await;
+ write_variant_partition(&store, 1, &["alphc", "alphd"], &[102, 103]).await;
+ write_test_metadata(&store, vec![0, 1], InvertedIndexParams::default()).await;
+ let cache = Arc::new(LanceCache::with_capacity(4096));
+ let index = InvertedIndex::load(store.clone(), None, cache.as_ref())
+ .await
+ .unwrap();
+
+ let params = FtsSearchParams::new()
+ .with_fuzziness(Some(1))
+ .with_max_expansions(3);
+ let tokens = Tokens::new(vec!["alphx".to_owned()], DocType::Text);
+
+ let expanded = index.expand_fuzzy_tokens(&tokens, ¶ms).unwrap();
+ let expanded_terms = (0..expanded.len())
+ .map(|idx| expanded.get_token(idx).to_owned())
+ .collect::>();
+ assert_eq!(
+ expanded_terms,
+ vec!["alpha".to_owned(), "alphb".to_owned(), "alphc".to_owned()],
+ "max_expansions must cap the whole query across partitions, \
+ in lexicographic order"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_fuzzy_results_independent_of_partition_shape() {
+ // The same four single-variant docs, laid out as one partition and
+ // as two. With a binding max_expansions the two shapes must still
+ // match the same documents with the same scores.
+ let single_dir = TempObjDir::default();
+ let single_store = Arc::new(LanceIndexStore::new(
+ ObjectStore::local().into(),
+ single_dir.clone(),
+ Arc::new(LanceCache::no_cache()),
+ ));
+ write_variant_partition(
+ &single_store,
+ 0,
+ &["alpha", "alphb", "alphc", "alphd"],
+ &[100, 101, 102, 103],
+ )
+ .await;
+ write_test_metadata(&single_store, vec![0], InvertedIndexParams::default()).await;
+
+ let split_dir = TempObjDir::default();
+ let split_store = Arc::new(LanceIndexStore::new(
+ ObjectStore::local().into(),
+ split_dir.clone(),
+ Arc::new(LanceCache::no_cache()),
+ ));
+ write_variant_partition(&split_store, 0, &["alpha", "alphb"], &[100, 101]).await;
+ write_variant_partition(&split_store, 1, &["alphc", "alphd"], &[102, 103]).await;
+ write_test_metadata(&split_store, vec![0, 1], InvertedIndexParams::default()).await;
+
+ let params = Arc::new(
+ FtsSearchParams::new()
+ .with_limit(Some(10))
+ .with_fuzziness(Some(1))
+ .with_max_expansions(3),
+ );
+
+ let mut results = Vec::new();
+ for store in [single_store, split_store] {
+ let cache = LanceCache::with_capacity(4096);
+ let index = InvertedIndex::load(store, None, &cache).await.unwrap();
+ let tokens = Arc::new(Tokens::new(vec!["alphx".to_owned()], DocType::Text));
+ let (row_ids, scores) = index
+ .bm25_search(
+ tokens,
+ params.clone(),
+ Operator::Or,
+ Arc::new(NoFilter),
+ Arc::new(NoOpMetricsCollector),
+ None,
+ )
+ .await
+ .unwrap();
+ let mut scored = row_ids.into_iter().zip(scores).collect::>();
+ scored.sort_unstable_by_key(|(row_id, _)| *row_id);
+ results.push(scored);
+ }
+
+ assert_eq!(
+ results[0]
+ .iter()
+ .map(|(row_id, _)| *row_id)
+ .collect::>(),
+ vec![100, 101, 102],
+ "a binding cap keeps the three lexicographically smallest variants"
+ );
+ assert_eq!(
+ results[0], results[1],
+ "fuzzy results must not depend on the partition shape"
+ );
+ }
+
#[tokio::test]
async fn test_fuzzy_and_scores_grouped_expansions_by_matched_token() {
let tmpdir = TempObjDir::default();
From 5e9c196f4b1524a696479d9c72e49e56eb35f5ca Mon Sep 17 00:00:00 2001
From: Xuanwo
Date: Tue, 7 Jul 2026 18:01:21 +0800
Subject: [PATCH 007/194] fix: build list FTS indexes as row documents (#7656)
## Summary
Close https://github.com/lance-format/lance/issues/5887
Build FTS indexes for list string columns as row-level documents instead
of flattening each list element into its own document.
This means `List`, `List`, and their `LargeList`
variants now treat a row as one document, with non-null list elements
contributing text fragments to that document. Token positions are
continuous across list elements, so phrase queries can match across
element boundaries.
The change intentionally does not add user parameters, persistent
metadata, legacy migration, or query-side deduplication. Old list
indexes keep their existing behavior.
## Context
Issue #5887 reports duplicate FTS results for list string columns
because the old index builder treated each list element as a separate
document with the same row id. Row-level indexing fixes the result
duplication and brings BM25 document statistics back to row-level
semantics for newly built indexes.
Follow-up issue #7654 tracks MemWAL in-memory FTS support for list
string columns.
## Validation
- `cargo fmt --all`
- `git diff --check`
- `cargo test -p lance-index flat_bm25_search`
- `cargo test -p lance-index test_worker_`
- `cargo test -p lance test_fts_list`
- `cargo test -p lance test_fts_index_with_`
- `cargo clippy -p lance-index --tests -- -D warnings`
- `cargo clippy -p lance --tests -- -D warnings`
---
.../src/scalar/inverted/builder.rs | 513 +++++++++---------
rust/lance-index/src/scalar/inverted/index.rs | 237 +++++++-
rust/lance/src/dataset/tests/dataset_index.rs | 163 ++++++
3 files changed, 627 insertions(+), 286 deletions(-)
diff --git a/rust/lance-index/src/scalar/inverted/builder.rs b/rust/lance-index/src/scalar/inverted/builder.rs
index ede00ad43ee..e3c2569f774 100644
--- a/rust/lance-index/src/scalar/inverted/builder.rs
+++ b/rust/lance-index/src/scalar/inverted/builder.rs
@@ -13,12 +13,12 @@ use crate::vector::graph::OrderedFloat;
use crate::{progress::IndexBuildProgress, progress::noop_progress};
use arrow::array::AsArray;
use arrow::datatypes;
-use arrow_array::{Array, BinaryArray, RecordBatch, UInt64Array};
+use arrow_array::{Array, BinaryArray, RecordBatch};
use arrow_schema::{DataType, Field, Schema, SchemaRef};
use bytes::Bytes;
-use datafusion::execution::{RecordBatchStream, SendableRecordBatchStream};
+use datafusion::execution::SendableRecordBatchStream;
use fst::Streamer;
-use futures::{Stream, StreamExt, TryStreamExt};
+use futures::{StreamExt, TryStreamExt};
use lance_arrow::json::JSON_EXT_NAME;
use lance_arrow::{ARROW_EXT_NAME_KEY, iter_str_array};
use lance_bitpacking::{BitPacker, BitPacker4x};
@@ -27,18 +27,16 @@ use lance_core::deepsize::DeepSizeOf;
use lance_core::error::LanceOptionExt;
use lance_core::utils::row_addr_remap::RowAddrRemap;
use lance_core::utils::tokio::{IO_CORE_RESERVATION, get_num_compute_intensive_cpus, spawn_cpu};
-use lance_core::{Error, ROW_ID, ROW_ID_FIELD, Result};
+use lance_core::{Error, ROW_ID, Result};
use lance_io::object_store::ObjectStore;
use lance_select::RowSetOps;
use object_store::path::Path;
use roaring::RoaringBitmap;
use smallvec::SmallVec;
use std::collections::HashMap;
-use std::pin::Pin;
use std::str::FromStr;
use std::sync::Arc;
use std::sync::LazyLock;
-use std::task::{Context, Poll};
use std::{fmt::Debug, sync::atomic::AtomicU64};
use tracing::instrument;
@@ -1264,6 +1262,11 @@ struct WorkerOutput {
tail_partition: Option,
}
+enum DocumentSource<'a> {
+ Text(&'a str),
+ StringList(&'a dyn Array),
+}
+
#[derive(Debug, Clone, Copy)]
struct IndexWorkerConfig {
with_position: bool,
@@ -1349,147 +1352,256 @@ impl IndexWorker {
async fn process_batch(&mut self, batch: RecordBatch) -> Result<()> {
let doc_col = batch.column(0);
- let doc_iter = iter_str_array(doc_col);
let row_id_col = batch[ROW_ID].as_primitive::();
- let docs = doc_iter
- .zip(row_id_col.values().iter())
- .filter_map(|(doc, row_id)| doc.map(|doc| (doc, *row_id)));
+ match doc_col.data_type() {
+ DataType::Utf8 | DataType::LargeUtf8 => {
+ let docs = iter_str_array(doc_col.as_ref())
+ .zip(row_id_col.values().iter())
+ .filter_map(|(doc, row_id)| doc.map(|doc| (doc, *row_id)));
+
+ for (doc, row_id) in docs {
+ self.process_document(row_id, DocumentSource::Text(doc), false)
+ .await?;
+ }
+ }
+ DataType::List(_) => {
+ self.process_string_list_batch::(doc_col, row_id_col)
+ .await?;
+ }
+ DataType::LargeList(_) => {
+ self.process_string_list_batch::(doc_col, row_id_col)
+ .await?;
+ }
+ data_type => {
+ return Err(Error::index(format!(
+ "expect data type String, LargeString, List(String), or LargeList(String) but got {}",
+ data_type
+ )));
+ }
+ }
+
+ Ok(())
+ }
+ async fn process_string_list_batch(
+ &mut self,
+ doc_col: &Arc,
+ row_id_col: &arrow_array::PrimitiveArray,
+ ) -> Result<()> {
+ let docs = doc_col.as_list::();
+ match docs.value_type() {
+ datatypes::DataType::Utf8 | datatypes::DataType::LargeUtf8 => {}
+ data_type => {
+ return Err(Error::index(format!(
+ "expect list item data type String or LargeString but got {}",
+ data_type
+ )));
+ }
+ }
+
+ for (doc, row_id) in docs.iter().zip(row_id_col.values().iter()) {
+ let Some(doc) = doc else {
+ continue;
+ };
+
+ self.process_document(*row_id, DocumentSource::StringList(doc.as_ref()), true)
+ .await?;
+ }
+
+ Ok(())
+ }
+
+ fn checked_token_position(row_id: u64, token_position: usize) -> Result {
+ u32::try_from(token_position).map_err(|_| {
+ Error::invalid_input(format!(
+ "token position overflow for row_id={row_id}: token_position={token_position}"
+ ))
+ })
+ }
+
+ fn materialize_string_list(elements: &dyn Array) -> String {
+ let mut doc = String::new();
+ for element in iter_str_array(elements).flatten() {
+ if !doc.is_empty() {
+ doc.push(' ');
+ }
+ doc.push_str(element);
+ }
+ doc
+ }
+
+ async fn process_document(
+ &mut self,
+ row_id: u64,
+ document: DocumentSource<'_>,
+ skip_empty_document: bool,
+ ) -> Result<()> {
let with_position = self.has_position();
- for (doc, row_id) in docs {
- let builder_was_empty = self.builder.docs.is_empty();
- let old_temporary_memory_size = self.temporary_memory_size();
- let old_token_memory_size = self.builder.tokens.memory_size() as u64;
- let doc_id = self.builder.docs.len() as u32;
- let mut token_num: u32 = 0;
- let mut posting_memory_delta = 0i64;
- if with_position {
+ let builder_was_empty = self.builder.docs.is_empty();
+ let old_temporary_memory_size = self.temporary_memory_size();
+ let old_token_memory_size = self.builder.tokens.memory_size() as u64;
+ let doc_id = self.builder.docs.len() as u32;
+ let mut token_num: u32 = 0;
+ let mut doc_length_bytes = 0usize;
+ let mut posting_memory_delta = 0i64;
+ if with_position {
+ {
if self.token_ids.capacity() < self.last_token_count {
self.token_ids
.reserve(self.last_token_count - self.token_ids.capacity());
}
self.token_ids.clear();
+ let tokenizer = &mut self.tokenizer;
let builder = &mut self.builder;
let token_ids = &mut self.token_ids;
let memory_size = &mut self.memory_size;
let posting_tail_codec = builder.posting_tail_codec;
- let mut token_stream = self.tokenizer.token_stream_for_doc(doc);
- while token_stream.advance() {
- let token = token_stream.token();
- let token_id = builder.tokens.get_or_add(&token.text);
- if token_id as usize == builder.posting_lists.len() {
- let old_posting_lists_overhead_size = (builder.posting_lists.capacity()
- * std::mem::size_of::())
- as u64;
- builder.posting_lists.push(
- PostingListBuilder::new_with_posting_tail_codec(
- true,
- posting_tail_codec,
- ),
- );
- let new_posting_lists_overhead_size = (builder.posting_lists.capacity()
- * std::mem::size_of::())
- as u64;
- Self::adjust_tracked_value(
- memory_size,
- old_posting_lists_overhead_size,
- new_posting_lists_overhead_size,
- );
+ let mut process_text = |text: &str| -> Result<()> {
+ doc_length_bytes += text.len();
+ let mut token_stream = tokenizer.token_stream_for_doc(text);
+ while token_stream.advance() {
+ let token = token_stream.token();
+ let position = Self::checked_token_position(row_id, token.position)?;
+ let token_id = builder.tokens.get_or_add(&token.text);
+ if token_id as usize == builder.posting_lists.len() {
+ let old_posting_lists_overhead_size = (builder.posting_lists.capacity()
+ * std::mem::size_of::())
+ as u64;
+ builder.posting_lists.push(
+ PostingListBuilder::new_with_posting_tail_codec(
+ true,
+ posting_tail_codec,
+ ),
+ );
+ let new_posting_lists_overhead_size = (builder.posting_lists.capacity()
+ * std::mem::size_of::())
+ as u64;
+ Self::adjust_tracked_value(
+ memory_size,
+ old_posting_lists_overhead_size,
+ new_posting_lists_overhead_size,
+ );
+ }
+ let posting_list = &mut builder.posting_lists[token_id as usize];
+ let old_posting_memory_size = posting_list.size();
+ if posting_list.add_occurrence(doc_id, position)? {
+ token_ids.push(token_id);
+ }
+ let new_posting_memory_size = posting_list.size();
+ posting_memory_delta +=
+ new_posting_memory_size as i64 - old_posting_memory_size as i64;
+ token_num += 1;
}
- let posting_list = &mut builder.posting_lists[token_id as usize];
- let old_posting_memory_size = posting_list.size();
- if posting_list.add_occurrence(doc_id, token.position as u32)? {
- token_ids.push(token_id);
+ Ok(())
+ };
+
+ match document {
+ DocumentSource::Text(doc) => {
+ process_text(doc)?;
+ }
+ DocumentSource::StringList(elements) => {
+ let doc = Self::materialize_string_list(elements);
+ process_text(&doc)?;
}
- let new_posting_memory_size = posting_list.size();
- posting_memory_delta +=
- new_posting_memory_size as i64 - old_posting_memory_size as i64;
- token_num += 1;
}
- } else {
+ }
+ } else {
+ {
if self.token_ids.capacity() < self.last_token_count {
self.token_ids
.reserve(self.last_token_count - self.token_ids.capacity());
}
self.token_ids.clear();
- let mut token_stream = self.tokenizer.token_stream_for_doc(doc);
- while token_stream.advance() {
- let token_id = self.builder.tokens.get_or_add(&token_stream.token().text);
- self.token_ids.push(token_id);
- token_num += 1;
- }
- }
- self.adjust_tracked_memory_size(
- old_token_memory_size,
- self.builder.tokens.memory_size() as u64,
- );
+ let tokenizer = &mut self.tokenizer;
+ let builder = &mut self.builder;
+ let token_ids = &mut self.token_ids;
+ let mut process_text = |text: &str| {
+ doc_length_bytes += text.len();
+ let mut token_stream = tokenizer.token_stream_for_doc(text);
+ while token_stream.advance() {
+ let token_id = builder.tokens.get_or_add(&token_stream.token().text);
+ token_ids.push(token_id);
+ token_num += 1;
+ }
+ };
- if !with_position {
- let old_posting_lists_overhead_size = self.posting_lists_overhead_size();
- self.builder
- .posting_lists
- .resize_with(self.builder.tokens.len(), || {
- PostingListBuilder::new_with_posting_tail_codec(
- false,
- self.builder.posting_tail_codec,
- )
- });
- let new_posting_lists_overhead_size = self.posting_lists_overhead_size();
- Self::adjust_tracked_value(
- &mut self.memory_size,
- old_posting_lists_overhead_size,
- new_posting_lists_overhead_size,
- );
+ match document {
+ DocumentSource::Text(doc) => process_text(doc),
+ DocumentSource::StringList(elements) => {
+ let doc = Self::materialize_string_list(elements);
+ process_text(&doc);
+ }
+ }
}
+ }
+ self.adjust_tracked_memory_size(
+ old_token_memory_size,
+ self.builder.tokens.memory_size() as u64,
+ );
- let old_doc_memory_size = self.builder.docs.memory_size() as u64;
- let appended_doc_id = self.builder.docs.append(row_id, token_num);
- debug_assert_eq!(appended_doc_id, doc_id);
+ if skip_empty_document && token_num == 0 {
+ self.last_token_count = 0;
+ self.trim_temporary_buffers();
self.adjust_tracked_memory_size(
- old_doc_memory_size,
- self.builder.docs.memory_size() as u64,
+ old_temporary_memory_size,
+ self.temporary_memory_size(),
);
- self.total_doc_length += doc.len();
+ return Ok(());
+ }
- if with_position {
- for &token_id in &self.token_ids {
- let (old_posting_memory_size, new_posting_memory_size) = {
- let posting_list = &mut self.builder.posting_lists[token_id as usize];
- let old_posting_memory_size = posting_list.size();
- posting_list.finish_open_doc(doc_id)?;
- let new_posting_memory_size = posting_list.size();
- (old_posting_memory_size, new_posting_memory_size)
- };
- posting_memory_delta +=
- new_posting_memory_size as i64 - old_posting_memory_size as i64;
- }
- Self::apply_delta(&mut self.memory_size, posting_memory_delta);
- } else if token_num > 0 {
- self.token_ids.sort_unstable();
- let mut iter = self.token_ids.iter();
- let mut current = *iter.next().unwrap();
- let mut count = 1u32;
- for &token_id in iter {
- if token_id == current {
- count += 1;
- continue;
- }
+ if !with_position {
+ let old_posting_lists_overhead_size = self.posting_lists_overhead_size();
+ self.builder
+ .posting_lists
+ .resize_with(self.builder.tokens.len(), || {
+ PostingListBuilder::new_with_posting_tail_codec(
+ false,
+ self.builder.posting_tail_codec,
+ )
+ });
+ let new_posting_lists_overhead_size = self.posting_lists_overhead_size();
+ Self::adjust_tracked_value(
+ &mut self.memory_size,
+ old_posting_lists_overhead_size,
+ new_posting_lists_overhead_size,
+ );
+ }
- let (old_posting_memory_size, new_posting_memory_size) = {
- let posting_list = &mut self.builder.posting_lists[current as usize];
- let old_posting_memory_size = posting_list.size();
- posting_list.add(doc_id, PositionRecorder::Count(count));
- let new_posting_memory_size = posting_list.size();
- (old_posting_memory_size, new_posting_memory_size)
- };
- posting_memory_delta +=
- new_posting_memory_size as i64 - old_posting_memory_size as i64;
+ let old_doc_memory_size = self.builder.docs.memory_size() as u64;
+ let appended_doc_id = self.builder.docs.append(row_id, token_num);
+ debug_assert_eq!(appended_doc_id, doc_id);
+ self.adjust_tracked_memory_size(
+ old_doc_memory_size,
+ self.builder.docs.memory_size() as u64,
+ );
+ self.total_doc_length += doc_length_bytes;
- current = token_id;
- count = 1;
+ if with_position {
+ for &token_id in &self.token_ids {
+ let (old_posting_memory_size, new_posting_memory_size) = {
+ let posting_list = &mut self.builder.posting_lists[token_id as usize];
+ let old_posting_memory_size = posting_list.size();
+ posting_list.finish_open_doc(doc_id)?;
+ let new_posting_memory_size = posting_list.size();
+ (old_posting_memory_size, new_posting_memory_size)
+ };
+ posting_memory_delta +=
+ new_posting_memory_size as i64 - old_posting_memory_size as i64;
+ }
+ Self::apply_delta(&mut self.memory_size, posting_memory_delta);
+ } else if token_num > 0 {
+ self.token_ids.sort_unstable();
+ let mut iter = self.token_ids.iter();
+ let mut current = *iter.next().unwrap();
+ let mut count = 1u32;
+ for &token_id in iter {
+ if token_id == current {
+ count += 1;
+ continue;
}
+
let (old_posting_memory_size, new_posting_memory_size) = {
let posting_list = &mut self.builder.posting_lists[current as usize];
let old_posting_memory_size = posting_list.size();
@@ -1499,27 +1611,35 @@ impl IndexWorker {
};
posting_memory_delta +=
new_posting_memory_size as i64 - old_posting_memory_size as i64;
- Self::apply_delta(&mut self.memory_size, posting_memory_delta);
- }
- self.last_token_count = self.token_ids.len();
- self.trim_temporary_buffers();
- self.adjust_tracked_memory_size(
- old_temporary_memory_size,
- self.temporary_memory_size(),
- );
- if self.builder.docs.len() == 1 && self.memory_size > self.worker_memory_limit_bytes {
- return Err(Error::invalid_input(format!(
- "single document row_id={} exceeds worker memory limit: {} > {} bytes",
- row_id, self.memory_size, self.worker_memory_limit_bytes
- )));
+ current = token_id;
+ count = 1;
}
+ let (old_posting_memory_size, new_posting_memory_size) = {
+ let posting_list = &mut self.builder.posting_lists[current as usize];
+ let old_posting_memory_size = posting_list.size();
+ posting_list.add(doc_id, PositionRecorder::Count(count));
+ let new_posting_memory_size = posting_list.size();
+ (old_posting_memory_size, new_posting_memory_size)
+ };
+ posting_memory_delta += new_posting_memory_size as i64 - old_posting_memory_size as i64;
+ Self::apply_delta(&mut self.memory_size, posting_memory_delta);
+ }
+ self.last_token_count = self.token_ids.len();
+ self.trim_temporary_buffers();
+ self.adjust_tracked_memory_size(old_temporary_memory_size, self.temporary_memory_size());
- if self.builder.docs.len() as u32 == u32::MAX
- || (!builder_was_empty && self.memory_size >= self.worker_memory_limit_bytes)
- {
- self.flush().await?;
- }
+ if self.builder.docs.len() == 1 && self.memory_size > self.worker_memory_limit_bytes {
+ return Err(Error::invalid_input(format!(
+ "single document row_id={} exceeds worker memory limit: {} > {} bytes",
+ row_id, self.memory_size, self.worker_memory_limit_bytes
+ )));
+ }
+
+ if self.builder.docs.len() as u32 == u32::MAX
+ || (!builder_was_empty && self.memory_size >= self.worker_memory_limit_bytes)
+ {
+ self.flush().await?;
}
Ok(())
@@ -1772,129 +1892,6 @@ fn inverted_list_schema_with_tail_codec_and_position_codec(
Arc::new(arrow_schema::Schema::new_with_metadata(fields, metadata))
}
-/// Flatten the string list stream into a string stream
-pub struct FlattenStream {
- /// Inner record batch stream with 2 columns:
- /// 1. doc_col: List(Utf8) or List(LargeUtf8)
- /// 2. row_id_col: UInt64
- inner: SendableRecordBatchStream,
- field_type: DataType,
- data_type: DataType,
-}
-
-impl FlattenStream {
- pub fn new(input: SendableRecordBatchStream) -> Self {
- let schema = input.schema();
- let field = schema.field(0);
- let data_type = match field.data_type() {
- DataType::List(f) if matches!(f.data_type(), DataType::Utf8) => DataType::Utf8,
- DataType::List(f) if matches!(f.data_type(), DataType::LargeUtf8) => {
- DataType::LargeUtf8
- }
- DataType::LargeList(f) if matches!(f.data_type(), DataType::Utf8) => DataType::Utf8,
- DataType::LargeList(f) if matches!(f.data_type(), DataType::LargeUtf8) => {
- DataType::LargeUtf8
- }
- _ => panic!(
- "expect data type List(Utf8) or List(LargeUtf8) but got {:?}",
- field.data_type()
- ),
- };
- Self {
- inner: input,
- field_type: field.data_type().clone(),
- data_type,
- }
- }
-}
-
-impl Stream for FlattenStream {
- type Item = datafusion_common::Result;
-
- fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll