From bdca8432858261d6e54a523886db5f9855054eac Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Tue, 15 Sep 2026 06:56:17 +0000 Subject: [PATCH] fix(rust): preserve compaction telemetry without plan cloning --- Cargo.lock | 1 + Cargo.toml | 1 + rust/lancedb/Cargo.toml | 1 + rust/lancedb/src/table/optimize.rs | 31 +++++++++++++++++++++++++----- 4 files changed, 29 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 160ee2253..eedcb2a0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5538,6 +5538,7 @@ dependencies = [ "tokenizers", "tokio", "tonic", + "tracing", "url", "urlencoding", "uuid", diff --git a/Cargo.toml b/Cargo.toml index 1068d48ee..fddb119e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -73,6 +73,7 @@ serde_json = "1" tempfile = "3.5.0" tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] } tonic = { version = "0.14", features = ["tls-native-roots", "tls-ring"] } +tracing = "0.1" uuid = { version = "1.7.0", features = ["v4", "v7"] } chrono = { version = "0.4", default-features = false, features = ["clock"] } diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index 9bc4c4a8c..e5deee910 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -55,6 +55,7 @@ moka = { workspace = true } pin-project = { workspace = true } tokio = { workspace = true } log.workspace = true +tracing.workspace = true async-trait = { workspace = true } bytes = { workspace = true } futures.workspace = true diff --git a/rust/lancedb/src/table/optimize.rs b/rust/lancedb/src/table/optimize.rs index 6d59e3692..e8f052269 100644 --- a/rust/lancedb/src/table/optimize.rs +++ b/rust/lancedb/src/table/optimize.rs @@ -6,7 +6,10 @@ //! This module contains the implementation of optimization operations that help //! maintain good performance for LanceDB tables. -use std::{collections::HashSet, sync::Arc}; +use std::{ + collections::HashSet, + sync::{Arc, Mutex}, +}; use arrow_schema::DataType; use chrono::{DateTime, Utc}; @@ -16,6 +19,7 @@ use lance::dataset::optimize::{ compact_files_with_planner, plan_compaction, }; use lance::index::{DatasetIndexExt, DatasetIndexInternalExt}; +use lance_core::utils::tracing::{DATASET_COMPACTING_EVENT, TRACE_DATASET_EVENTS}; use lance_index::IndexType; use lance_index::metrics::NoOpMetricsCollector; use lance_index::optimize::OptimizeOptions; @@ -29,14 +33,30 @@ pub use lance::dataset::optimize::CompactionOptions; const MAX_ARROW_FIXED_SIZE_LIST_CHILD_INDEX: u64 = u32::MAX as u64; struct PrecomputedCompactionPlanner { - plan: CompactionPlan, + plan: Mutex>, +} + +impl PrecomputedCompactionPlanner { + fn new(plan: CompactionPlan) -> Self { + Self { + plan: Mutex::new(Some(plan)), + } + } } #[async_trait::async_trait] impl CompactionPlanner for PrecomputedCompactionPlanner { async fn plan(&self, dataset: &lance::Dataset) -> lance::Result { - debug_assert_eq!(dataset.manifest().version, self.plan.read_version()); - Ok(self.plan.clone()) + let plan = self + .plan + .lock() + .map_err(|_| lance_core::Error::internal("precomputed compaction plan lock poisoned"))? + .take() + .ok_or_else(|| { + lance_core::Error::internal("precomputed compaction plan was already consumed") + })?; + debug_assert_eq!(dataset.manifest().version, plan.read_version()); + Ok(plan) } } @@ -331,8 +351,9 @@ pub(crate) async fn compact_files_impl( ) -> Result { table.dataset.ensure_mutable()?; let mut dataset = (*table.dataset.get().await?).clone(); + tracing::info!(target: TRACE_DATASET_EVENTS, event=DATASET_COMPACTING_EVENT, uri = dataset.uri()); let plan = validate_sq_index_remapping(&dataset, &options, remap_options.is_some()).await?; - let planner = PrecomputedCompactionPlanner { plan }; + let planner = PrecomputedCompactionPlanner::new(plan); let metrics = compact_files_with_planner(&mut dataset, remap_options, &planner).await?; table.dataset.update(dataset); Ok(metrics)