mirror of
https://github.com/lancedb/lancedb.git
synced 2026-08-28 08:58:41 +00:00
Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 74c003af59 | |||
| 695e7ae956 | |||
| 67532dc4ce | |||
| 04836120d4 | |||
| 160b72bbd9 | |||
| 53c6495a32 |
@@ -296,18 +296,16 @@ jobs:
|
||||
cargo update -p aws-types --precise 1.3.9
|
||||
cargo update -p aws-sigv4 --precise 1.3.5
|
||||
cargo update -p aws-credential-types --precise 1.2.8
|
||||
# aws-smithy-checksums must stay at or above 0.63.13: OpenDAL's S3
|
||||
# service needs crc-fast ~1.9, and older releases pin it to ~1.3.
|
||||
cargo update -p aws-smithy-checksums --precise 0.63.13
|
||||
cargo update -p aws-smithy-checksums --precise 0.63.9
|
||||
cargo update -p aws-smithy-runtime --precise 1.9.3
|
||||
cargo update -p aws-smithy-http --precise 0.62.6
|
||||
cargo update -p aws-smithy-eventstream --precise 0.60.14
|
||||
cargo update -p aws-smithy-http --precise 0.62.4
|
||||
cargo update -p aws-smithy-eventstream --precise 0.60.12
|
||||
cargo update -p aws-smithy-http-client --precise 1.1.3
|
||||
cargo update -p aws-smithy-observability --precise 0.1.4
|
||||
cargo update -p aws-smithy-query --precise 0.60.8
|
||||
cargo update -p aws-smithy-runtime-api --precise 1.9.3
|
||||
cargo update -p aws-smithy-async --precise 1.2.7
|
||||
cargo update -p aws-smithy-types --precise 1.3.6
|
||||
cargo update -p aws-smithy-runtime-api --precise 1.9.1
|
||||
cargo update -p aws-smithy-async --precise 1.2.6
|
||||
cargo update -p aws-smithy-types --precise 1.3.5
|
||||
cargo update -p aws-smithy-xml --precise 0.60.11
|
||||
cargo update -p home --precise 0.5.9
|
||||
- name: cargo +${{ matrix.msrv }} check
|
||||
|
||||
@@ -152,54 +152,3 @@ Please consider the following when reviewing code contributions.
|
||||
### Documentation
|
||||
* New features must include updates to the rust documentation comments. Link to
|
||||
relevant structs and methods to increase the value of documentation.
|
||||
|
||||
## Cursor Cloud specific instructions
|
||||
|
||||
The VM snapshot already has the Rust `1.97.0` toolchain (auto-selected by
|
||||
`rust-toolchain.toml`), `protoc`, `uv` (on `PATH` via `~/.bashrc`), the Rust
|
||||
debug build artifacts, the Python editable extension, and `nodejs/node_modules`.
|
||||
The startup update script only refreshes dependencies (`uv sync` for Python and
|
||||
`pnpm install` for Node); it deliberately does NOT rebuild the native
|
||||
extensions. After changing Rust or PyO3/napi binding code you must rebuild the
|
||||
affected binding yourself (see per-binding rebuild commands below).
|
||||
|
||||
Non-obvious caveats discovered during setup:
|
||||
|
||||
* The documented Python bootstrap `uv run --extra tests --extra dev maturin
|
||||
develop --extras tests,dev` does not work as-is here: `maturin` is not
|
||||
installed as a CLI in the uv environment, and `maturin develop --extras`
|
||||
runs its own dependency resolution that cannot find the prerelease
|
||||
`pylance==9.0.0rc1` (it lacks the extra package index that `uv` uses via
|
||||
`uv.lock`). Because `uv run --extra tests --extra dev` already installs those
|
||||
extras, the working command is:
|
||||
`cd python && uv run --extra tests --extra dev --with maturin maturin develop`
|
||||
(note: `--with maturin`, and no `--extras`). This is the Python binding
|
||||
rebuild command.
|
||||
* Rust core, the Python extension (maturin), and the Node addon (napi) all
|
||||
compile into the SHARED `/workspace/target`. Cargo feature unification differs
|
||||
between `maturin develop` and `pnpm build`, so alternating between building
|
||||
the Python and Node bindings forces a full recompile of shared crates
|
||||
(`lancedb`, `datafusion`, `lance-*`) — roughly 6-7 min each way on this
|
||||
4-core VM. Build one binding at a time to avoid the churn.
|
||||
* The `_lancedb` release build (triggered when `uv run`/`uv sync` installs the
|
||||
`lancedb` project itself) uses `lto = "fat"` + `opt-level = 3`, needs ~11 GB
|
||||
RAM, and takes ~20 min cold on this VM. To avoid it, the update script uses
|
||||
`uv sync --no-install-project --inexact` (the `--inexact` flag is required so
|
||||
the sync does not uninstall the editable extension). Prefer the debug
|
||||
`maturin develop` (~6 min cold, seconds when warm) for iteration.
|
||||
* `cargo check` only produces metadata, so the first `cargo run --example ...`
|
||||
or `cargo test` after a check triggers a large codegen/link compile.
|
||||
* Node binding rebuild: `cd nodejs && pnpm build` (napi debug build + `tsc`).
|
||||
The native addon lands at `nodejs/dist/lancedb.linux-x64-gnu.node`.
|
||||
|
||||
Verified working (local backend, no cloud credentials needed):
|
||||
|
||||
* Rust: `cargo check/clippy --features remote --tests --examples`,
|
||||
`cargo test --features remote -p lancedb --lib`, `cargo run --features remote
|
||||
--example simple`.
|
||||
* Python: `cd python && uv run --extra tests pytest python/tests/test_table.py`,
|
||||
`uv run --directory python --extra dev ruff check python`.
|
||||
* Node: `cd nodejs && pnpm lint`, `pnpm test __test__/connection.test.ts`.
|
||||
|
||||
Java (`java/`) is optional; its integration tests need LanceDB Cloud
|
||||
credentials (`LANCEDB_DB`, `LANCEDB_API_KEY`) and were not set up here.
|
||||
|
||||
Generated
+233
-257
File diff suppressed because it is too large
Load Diff
+14
-14
@@ -13,20 +13,20 @@ categories = ["database-implementations"]
|
||||
rust-version = "1.91.0"
|
||||
|
||||
[workspace.dependencies]
|
||||
lance = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=11.0.0-beta.2", default-features = false, "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=11.0.0-beta.2", "tag" = "v11.0.0-beta.2", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance = { "version" = "=10.1.0-beta.1", default-features = false, "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-core = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datagen = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-file = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-io = { "version" = "=10.1.0-beta.1", default-features = false, "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-index = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-linalg = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-namespace-impls = { "version" = "=10.1.0-beta.1", default-features = false, "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-table = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-testing = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-datafusion = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-encoding = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "git" = "https://github.com/lance-format/lance.git" }
|
||||
lance-arrow = { "version" = "=10.1.0-beta.1", "tag" = "v10.1.0-beta.1", "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 }
|
||||
|
||||
+1
-1
@@ -28,7 +28,7 @@
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<arrow.version>15.0.0</arrow.version>
|
||||
<lance-core.version>11.0.0-beta.2</lance-core.version>
|
||||
<lance-core.version>10.1.0-beta.1</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>
|
||||
|
||||
@@ -49,8 +49,8 @@ lance-namespace = { workspace = true }
|
||||
lance-namespace-impls = { workspace = true }
|
||||
metrics = { workspace = true, optional = true }
|
||||
metrics-util = { workspace = true, optional = true }
|
||||
# Pin the GooseFS SDK to the version required by Lance's OpenDAL dependency.
|
||||
goosefs-sdk = { version = "=0.1.9", optional = true }
|
||||
# Pin the transitive GooseFS SDK until the 0.1.6 compile break is fixed upstream.
|
||||
goosefs-sdk = { version = "=0.1.5", optional = true }
|
||||
moka = { workspace = true }
|
||||
pin-project = { workspace = true }
|
||||
tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] }
|
||||
|
||||
@@ -17,7 +17,7 @@ use arrow_array::builder::LargeBinaryBuilder;
|
||||
use arrow_schema::{DataType, Field, Schema};
|
||||
use lance::dataset::{BlobRangeRequest as LanceBlobRangeRequest, Dataset, WriteParams};
|
||||
use lance_arrow::FieldExt;
|
||||
use lance_file::version::LanceFileVersion;
|
||||
use lance_encoding::version::LanceFileVersion;
|
||||
use lance_io::object_store::ObjectStore;
|
||||
use object_store::path::Path;
|
||||
|
||||
|
||||
@@ -34,7 +34,7 @@ use crate::remote::{
|
||||
db::{OPT_REMOTE_API_KEY, OPT_REMOTE_HOST_OVERRIDE, OPT_REMOTE_REGION},
|
||||
};
|
||||
use lance::io::ObjectStoreParams;
|
||||
pub use lance_file::version::LanceFileVersion;
|
||||
pub use lance_encoding::version::LanceFileVersion;
|
||||
#[cfg(feature = "remote")]
|
||||
use lance_io::object_store::StorageOptions;
|
||||
use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
|
||||
|
||||
@@ -12,7 +12,7 @@ use lance::dataset::refs::Ref;
|
||||
use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder};
|
||||
use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore};
|
||||
use lance_datafusion::utils::StreamingWriteSource;
|
||||
use lance_file::version::LanceFileVersion;
|
||||
use lance_encoding::version::LanceFileVersion;
|
||||
use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider};
|
||||
use lance_table::io::commit::commit_handler_from_url;
|
||||
use object_store::local::LocalFileSystem;
|
||||
|
||||
@@ -201,7 +201,7 @@ impl LanceNamespaceDatabase {
|
||||
&self,
|
||||
request: &DbCreateTableRequest,
|
||||
) -> Result<(
|
||||
Option<lance_file::version::LanceFileVersion>,
|
||||
Option<lance_encoding::version::LanceFileVersion>,
|
||||
Option<bool>,
|
||||
Option<bool>,
|
||||
)> {
|
||||
@@ -214,7 +214,7 @@ impl LanceNamespaceDatabase {
|
||||
|
||||
let storage_version_override = storage_options
|
||||
.and_then(|opts| opts.get(OPT_NEW_TABLE_STORAGE_VERSION))
|
||||
.map(|s| s.parse::<lance_file::version::LanceFileVersion>())
|
||||
.map(|s| s.parse::<lance_encoding::version::LanceFileVersion>())
|
||||
.transpose()?;
|
||||
|
||||
let v2_manifest_override = storage_options
|
||||
|
||||
@@ -2942,7 +2942,7 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
}
|
||||
|
||||
#[derive(Serialize, Clone, Debug)]
|
||||
pub struct MergeInsertRequest {
|
||||
pub(crate) struct MergeInsertRequest {
|
||||
on: String,
|
||||
when_matched_update_all: bool,
|
||||
when_matched_update_all_filt: Option<String>,
|
||||
@@ -5907,18 +5907,16 @@ mod tests {
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Positions are relative to the first retained token, so dropping the
|
||||
// leading "hello" stop word does not shift the remaining tokens.
|
||||
assert_eq!(
|
||||
tokens,
|
||||
vec![
|
||||
FtsToken {
|
||||
text: "こんにちは".to_string(),
|
||||
position: 0,
|
||||
position: 1,
|
||||
},
|
||||
FtsToken {
|
||||
text: "世界".to_string(),
|
||||
position: 1,
|
||||
position: 2,
|
||||
},
|
||||
]
|
||||
);
|
||||
|
||||
@@ -90,7 +90,7 @@ struct RemoteBlobState {
|
||||
|
||||
/// Seekable Cloud blob handle over HTTP Range.
|
||||
#[derive(Debug)]
|
||||
pub struct RemoteBlobFile {
|
||||
pub(crate) struct RemoteBlobFile {
|
||||
requester: Arc<dyn BlobRangeRequester>,
|
||||
state: Mutex<RemoteBlobState>,
|
||||
closed: AtomicBool,
|
||||
|
||||
@@ -33,7 +33,7 @@ use crate::table::{AddResult, MergeResult};
|
||||
/// same Arrow-IPC streaming body and error side-channel; only the target
|
||||
/// endpoint, query parameters, and parsed result type differ.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum WriteOp {
|
||||
pub(crate) enum WriteOp {
|
||||
/// `add`: stream to `/v1/table/{id}/insert/`, optionally overwriting.
|
||||
Insert { overwrite: bool },
|
||||
/// `merge_insert`: stream to `/v1/table/{id}/merge_insert/` with the merge
|
||||
@@ -49,7 +49,7 @@ pub enum WriteOp {
|
||||
/// The parsed server response for a completed write, discriminated by the
|
||||
/// operation that produced it.
|
||||
#[derive(Debug, Clone)]
|
||||
pub enum WriteResult {
|
||||
pub(crate) enum WriteResult {
|
||||
Add(AddResult),
|
||||
Merge(MergeResult),
|
||||
}
|
||||
|
||||
@@ -6,19 +6,162 @@
|
||||
//! This module contains the implementation of optimization operations that help
|
||||
//! maintain good performance for LanceDB tables.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::{collections::HashSet, sync::Arc};
|
||||
|
||||
use arrow_schema::DataType;
|
||||
use lance::dataset::cleanup::RemovalStats;
|
||||
use lance::dataset::optimize::{CompactionMetrics, IndexRemapperOptions, compact_files};
|
||||
use lance::index::DatasetIndexExt;
|
||||
use lance::dataset::optimize::{
|
||||
CompactionMetrics, CompactionPlan, CompactionPlanner, IndexRemapperOptions,
|
||||
compact_files_with_planner, plan_compaction,
|
||||
};
|
||||
use lance::index::{DatasetIndexExt, DatasetIndexInternalExt};
|
||||
use lance_index::IndexType;
|
||||
use lance_index::metrics::NoOpMetricsCollector;
|
||||
use lance_index::optimize::OptimizeOptions;
|
||||
use log::info;
|
||||
use log::{debug, info};
|
||||
|
||||
use super::NativeTable;
|
||||
use crate::error::{Error, Result};
|
||||
pub use chrono::Duration;
|
||||
pub use lance::dataset::optimize::CompactionOptions;
|
||||
|
||||
use super::NativeTable;
|
||||
use crate::error::Result;
|
||||
const MAX_ARROW_FIXED_SIZE_LIST_CHILD_INDEX: u64 = u32::MAX as u64;
|
||||
|
||||
struct PrecomputedCompactionPlanner {
|
||||
plan: CompactionPlan,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl CompactionPlanner for PrecomputedCompactionPlanner {
|
||||
async fn plan(&self, dataset: &lance::Dataset) -> lance::Result<CompactionPlan> {
|
||||
debug_assert_eq!(dataset.manifest().version, self.plan.read_version());
|
||||
Ok(self.plan.clone())
|
||||
}
|
||||
}
|
||||
|
||||
fn sq_vector_dimension(data_type: &DataType) -> Option<i32> {
|
||||
match data_type {
|
||||
DataType::FixedSizeList(_, dimension) => Some(*dimension),
|
||||
DataType::List(field) => match field.data_type() {
|
||||
DataType::FixedSizeList(_, dimension) => Some(*dimension),
|
||||
_ => None,
|
||||
},
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn segment_touches_fragments(
|
||||
fragment_ids: Option<impl IntoIterator<Item = u32>>,
|
||||
affected_fragments: &HashSet<u64>,
|
||||
) -> bool {
|
||||
fragment_ids.is_none_or(|fragment_ids| {
|
||||
fragment_ids
|
||||
.into_iter()
|
||||
.any(|fragment| affected_fragments.contains(&(fragment as u64)))
|
||||
})
|
||||
}
|
||||
|
||||
fn oversized_sq_partition(
|
||||
partition_sizes: impl IntoIterator<Item = u64>,
|
||||
dimension: u64,
|
||||
) -> Option<u64> {
|
||||
partition_sizes.into_iter().find(|partition_size| {
|
||||
partition_size
|
||||
.checked_mul(dimension)
|
||||
.is_none_or(|child_len| child_len > MAX_ARROW_FIXED_SIZE_LIST_CHILD_INDEX)
|
||||
})
|
||||
}
|
||||
|
||||
async fn validate_sq_index_remapping(
|
||||
dataset: &lance::Dataset,
|
||||
options: &CompactionOptions,
|
||||
has_custom_remapper: bool,
|
||||
) -> Result<CompactionPlan> {
|
||||
let plan = plan_compaction(dataset, options).await?;
|
||||
if options.defer_index_remap || has_custom_remapper || dataset.manifest().uses_stable_row_ids()
|
||||
{
|
||||
return Ok(plan);
|
||||
}
|
||||
|
||||
if plan.tasks.is_empty() {
|
||||
return Ok(plan);
|
||||
}
|
||||
let affected_fragments: HashSet<u64> = plan
|
||||
.tasks
|
||||
.iter()
|
||||
.flat_map(|task| task.fragments.iter().map(|fragment| fragment.id))
|
||||
.collect();
|
||||
|
||||
for segment in dataset.load_indices().await?.iter() {
|
||||
if !segment_touches_fragments(
|
||||
segment
|
||||
.fragment_bitmap
|
||||
.as_ref()
|
||||
.map(|fragment_bitmap| fragment_bitmap.iter()),
|
||||
&affected_fragments,
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
let Some(field_id) = segment.fields.first() else {
|
||||
continue;
|
||||
};
|
||||
let Some(field) = dataset.schema().field_by_id(*field_id) else {
|
||||
continue;
|
||||
};
|
||||
let data_type = field.data_type();
|
||||
let Some(dimension) = sq_vector_dimension(&data_type) else {
|
||||
continue;
|
||||
};
|
||||
let dimension = u64::try_from(dimension).map_err(|_| Error::InvalidInput {
|
||||
message: format!(
|
||||
"SQ index '{}' has an invalid vector dimension of {}",
|
||||
segment.name, dimension
|
||||
),
|
||||
})?;
|
||||
|
||||
let field_path = dataset.schema().field_path(*field_id)?;
|
||||
let vector_index = match dataset
|
||||
.open_vector_index(&field_path, &segment.uuid, &NoOpMetricsCollector)
|
||||
.await
|
||||
{
|
||||
Ok(vector_index) => vector_index,
|
||||
Err(error) => {
|
||||
// The default remapper also treats an index it cannot open as
|
||||
// unusable and drops it, so it cannot reach the SQ take kernel.
|
||||
debug!(
|
||||
"Skipping remap preflight for index segment {} because it could not be opened: {}",
|
||||
segment.uuid, error
|
||||
);
|
||||
continue;
|
||||
}
|
||||
};
|
||||
if !matches!(
|
||||
vector_index.index_type(),
|
||||
IndexType::IvfSq | IndexType::IvfHnswSq
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
|
||||
let partition_sizes = (0..vector_index.total_partitions())
|
||||
.map(|partition_id| vector_index.partition_size(partition_id) as u64);
|
||||
if let Some(partition_size) = oversized_sq_partition(partition_sizes, dimension) {
|
||||
let max_partition_size = MAX_ARROW_FIXED_SIZE_LIST_CHILD_INDEX / dimension;
|
||||
return Err(Error::InvalidInput {
|
||||
message: format!(
|
||||
"Cannot compact table because SQ index '{}' segment {} has a partition with {} vectors of dimension {}. Arrow's fixed-size-list take kernel cannot remap partitions whose child array exceeds {} values. Recreate the index with default IVF partitioning, or enough partitions to keep each partition at or below {} vectors, before compacting.",
|
||||
segment.name,
|
||||
segment.uuid,
|
||||
partition_size,
|
||||
dimension,
|
||||
MAX_ARROW_FIXED_SIZE_LIST_CHILD_INDEX,
|
||||
max_partition_size,
|
||||
),
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Ok(plan)
|
||||
}
|
||||
|
||||
/// Optimize the dataset.
|
||||
///
|
||||
@@ -144,7 +287,7 @@ pub(crate) async fn cleanup_old_versions(
|
||||
/// This can be run after making several small appends to optimize the table
|
||||
/// for faster reads.
|
||||
///
|
||||
/// This calls into [lance::dataset::optimize::compact_files].
|
||||
/// This calls into [lance::dataset::optimize::compact_files_with_planner].
|
||||
pub(crate) async fn compact_files_impl(
|
||||
table: &NativeTable,
|
||||
options: CompactionOptions,
|
||||
@@ -152,7 +295,9 @@ pub(crate) async fn compact_files_impl(
|
||||
) -> Result<CompactionMetrics> {
|
||||
table.dataset.ensure_mutable()?;
|
||||
let mut dataset = (*table.dataset.get().await?).clone();
|
||||
let metrics = compact_files(&mut dataset, options, remap_options).await?;
|
||||
let plan = validate_sq_index_remapping(&dataset, &options, remap_options.is_some()).await?;
|
||||
let planner = PrecomputedCompactionPlanner { plan };
|
||||
let metrics = compact_files_with_planner(&mut dataset, remap_options, &planner).await?;
|
||||
table.dataset.update(dataset);
|
||||
Ok(metrics)
|
||||
}
|
||||
@@ -214,13 +359,14 @@ pub(crate) async fn execute_optimize(
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{oversized_sq_partition, segment_touches_fragments, sq_vector_dimension};
|
||||
use arrow_array::{
|
||||
Array, FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray,
|
||||
};
|
||||
use arrow_schema::{DataType, Field, Schema};
|
||||
use lance_arrow::FixedSizeListArrayExt;
|
||||
use rstest::rstest;
|
||||
use std::sync::Arc;
|
||||
use std::{collections::HashSet, sync::Arc};
|
||||
|
||||
use crate::connect;
|
||||
use crate::database::listing::OPT_NEW_TABLE_ENABLE_STABLE_ROW_IDS;
|
||||
@@ -230,6 +376,172 @@ mod tests {
|
||||
use crate::table::{CompactionOptions, OptimizeAction, OptimizeStats};
|
||||
use futures::TryStreamExt;
|
||||
|
||||
/// Regression test for https://github.com/lancedb/lancedb/issues/2866.
|
||||
#[test]
|
||||
fn test_detect_oversized_sq_partition() {
|
||||
const DIMENSION: u64 = 4095;
|
||||
let safe_size = u32::MAX as u64 / DIMENSION;
|
||||
|
||||
assert_eq!(
|
||||
oversized_sq_partition([safe_size, safe_size + 1], DIMENSION),
|
||||
Some(safe_size + 1)
|
||||
);
|
||||
assert_eq!(
|
||||
oversized_sq_partition([safe_size / 2, safe_size / 2 + 1], DIMENSION),
|
||||
None
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_segment_touches_fragments_includes_unknown_coverage() {
|
||||
let affected_fragments = HashSet::from([7]);
|
||||
|
||||
assert!(segment_touches_fragments(
|
||||
Some([7_u32]),
|
||||
&affected_fragments
|
||||
));
|
||||
assert!(!segment_touches_fragments(
|
||||
Some([8_u32]),
|
||||
&affected_fragments
|
||||
));
|
||||
assert!(segment_touches_fragments(
|
||||
None::<[u32; 0]>,
|
||||
&affected_fragments
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_sq_vector_dimension_includes_multivectors() {
|
||||
const DIMENSION: i32 = 4095;
|
||||
let vector = DataType::FixedSizeList(
|
||||
Arc::new(Field::new("item", DataType::Float32, false)),
|
||||
DIMENSION,
|
||||
);
|
||||
let multivector = DataType::List(Arc::new(Field::new("item", vector.clone(), false)));
|
||||
|
||||
assert_eq!(sq_vector_dimension(&vector), Some(DIMENSION));
|
||||
assert_eq!(sq_vector_dimension(&multivector), Some(DIMENSION));
|
||||
assert_eq!(sq_vector_dimension(&DataType::Float32), None);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_compact_legacy_vector_index_with_unknown_fragment_coverage() {
|
||||
use lance::Dataset;
|
||||
use lance::index::vector::{IndexFileVersion, VectorIndexParams};
|
||||
use lance::index::{DatasetIndexExt, DatasetIndexInternalExt};
|
||||
use lance_index::IndexType;
|
||||
use lance_linalg::distance::MetricType;
|
||||
use lance_table::io::commit::write_manifest_file_to_path;
|
||||
use object_store::ObjectStoreExt;
|
||||
|
||||
const INDEX_NAME: &str = "legacy_ivf_pq";
|
||||
const ROWS: i32 = 128;
|
||||
const DIMENSION: i32 = 8;
|
||||
|
||||
let tmpdir = tempfile::tempdir().unwrap();
|
||||
let conn = connect(tmpdir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let vectors = FixedSizeListArray::try_new_from_values(
|
||||
Float32Array::from_iter_values((0..ROWS).flat_map(|row| {
|
||||
(0..DIMENSION).map(move |offset| (row * DIMENSION + offset) as f32)
|
||||
})),
|
||||
DIMENSION,
|
||||
)
|
||||
.unwrap();
|
||||
let schema = Arc::new(Schema::new(vec![Field::new(
|
||||
"vector",
|
||||
vectors.data_type().clone(),
|
||||
false,
|
||||
)]));
|
||||
let batch = RecordBatch::try_new(schema, vec![Arc::new(vectors)]).unwrap();
|
||||
let table = conn
|
||||
.create_table("test_legacy_vector_compact", batch.clone())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
table.add(batch.clone()).execute().await.unwrap();
|
||||
table.add(batch).execute().await.unwrap();
|
||||
let mut dataset = (*table.dataset().unwrap().get().await.unwrap()).clone();
|
||||
let mut params = VectorIndexParams::ivf_pq(1, 8, 1, MetricType::L2, 10);
|
||||
params.version(IndexFileVersion::Legacy);
|
||||
dataset
|
||||
.create_index(
|
||||
&["vector"],
|
||||
IndexType::Vector,
|
||||
Some(INDEX_NAME.to_string()),
|
||||
¶ms,
|
||||
true,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
table.dataset().unwrap().update(dataset);
|
||||
|
||||
let dataset = table.dataset().unwrap().get().await.unwrap();
|
||||
let mut segments = dataset.load_indices_by_name(INDEX_NAME).await.unwrap();
|
||||
assert_eq!(segments.len(), 1);
|
||||
let original = segments.pop().unwrap();
|
||||
let field_path = dataset.schema().field_path(original.fields[0]).unwrap();
|
||||
let vector_index = dataset
|
||||
.open_vector_index(
|
||||
&field_path,
|
||||
&original.uuid,
|
||||
&lance_index::metrics::NoOpMetricsCollector,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(vector_index.index_type(), IndexType::IvfPq);
|
||||
assert_eq!(
|
||||
vector_index.statistics().unwrap()["index_file_version"],
|
||||
"Legacy"
|
||||
);
|
||||
let mut legacy = original.clone();
|
||||
legacy.fragment_bitmap = None;
|
||||
let object_store = dataset.object_store(None).await.unwrap();
|
||||
let mut manifest = dataset.manifest().clone();
|
||||
manifest.index_section = None;
|
||||
manifest.transaction_section = None;
|
||||
let manifest_path = dataset.manifest_location().path.clone();
|
||||
object_store.inner.delete(&manifest_path).await.unwrap();
|
||||
write_manifest_file_to_path(
|
||||
object_store.as_ref(),
|
||||
&mut manifest,
|
||||
Some(vec![legacy]),
|
||||
&manifest_path,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let legacy_dataset = Dataset::open(dataset.uri()).await.unwrap();
|
||||
table.dataset().unwrap().update(legacy_dataset);
|
||||
|
||||
let dataset = table.dataset().unwrap().get().await.unwrap();
|
||||
let description_error = dataset
|
||||
.describe_indices(None)
|
||||
.await
|
||||
.err()
|
||||
.expect("legacy coverage should be unknown to index descriptions");
|
||||
assert!(
|
||||
description_error
|
||||
.to_string()
|
||||
.contains("Fragment bitmap is required")
|
||||
);
|
||||
|
||||
let stats = table
|
||||
.optimize(OptimizeAction::Compact {
|
||||
options: CompactionOptions {
|
||||
target_rows_per_fragment: 1_000,
|
||||
..Default::default()
|
||||
},
|
||||
remap_options: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(stats.compaction.unwrap().fragments_removed > 0);
|
||||
assert_eq!(table.count_rows(None).await.unwrap(), ROWS as usize * 3);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_optimize_compact_simple() {
|
||||
let conn = connect("memory://").execute().await.unwrap();
|
||||
|
||||
@@ -10,7 +10,7 @@ use arrow_array::{
|
||||
use arrow_schema::{DataType, Field, Fields, Schema};
|
||||
use futures::TryStreamExt;
|
||||
use lance::Dataset;
|
||||
use lance_file::version::LanceFileVersion;
|
||||
use lance_encoding::version::LanceFileVersion;
|
||||
use lancedb::{
|
||||
Connection, Error, Result, Table,
|
||||
blob::{BlobRangeRequest, blob},
|
||||
|
||||
Reference in New Issue
Block a user