From 3c9220ea1a6e178f877ee4a31463d48055cd4452 Mon Sep 17 00:00:00 2001 From: Ryan Mack Date: Tue, 18 Aug 2026 08:43:07 -0400 Subject: [PATCH 1/3] Build the fragment-reuse index open path on RowAddrRemap Forward-port of rerun-io/lance#38 to release-9. `RowAddrRemap` is already here (upstream lance-format/lance#7237), but `open_frag_reuse_index` was never converted to it, on this branch or on upstream main and v10.0. It built a `HashMap>` per reuse version with one entry per remapped row, on every index open, cached, so readers pay it. Measured on a production payload: 676,592,102 entries, 88 MB on disk becoming 26.8 GB resident and 40.5 GB peak, 144 s to build. It OOMs a 60 GiB pod. Against the same payload, `RowAddrRemap::compact` builds in 0.1 s and 0.27 GB peak, and agrees with the map on all 676,595,694 addresses probed. The only differences are 3,592 offsets past a fragment's `physical_rows`, where the map says absent and compact says deleted; that is compact's documented behaviour and those addresses are not rows. This is the residual delta for redap 0.16: everything in #38 except the backport, which this branch already has. Three fork-local additions to the upstream module, marked as such above the test module: `Debug` and `DeepSizeOf` for `RowAddrRemap`, plus the `num_groups`/`num_fragments` accessors those use. `FragReuseIndex` needs both; upstream does not yet, because it still holds hashmaps there. `remap_column_index` composed a map by enumerating keys, and the compact form deliberately cannot enumerate: it stores per-fragment bitmaps and treats any unlisted offset in a rewritten fragment as deleted, so its key set is not finite. That path now rebuilds the per-row maps from the details. Still O(rows), but only tests reach it in-tree and no redap caller does. Co-Authored-By: Claude Opus 5 (1M context) --- rust/lance-core/src/utils/row_addr_remap.rs | 58 +++++++++++++++++++ rust/lance-index/src/vector/bq/storage.rs | 2 +- .../src/system_index/frag_reuse.rs | 38 +++++++++--- rust/lance/src/dataset/optimize/remapping.rs | 19 +++++- rust/lance/src/index/frag_reuse.rs | 36 +++++++----- 5 files changed, 127 insertions(+), 26 deletions(-) diff --git a/rust/lance-core/src/utils/row_addr_remap.rs b/rust/lance-core/src/utils/row_addr_remap.rs index 6f5a6f2aae5..c21bc11a1b0 100644 --- a/rust/lance-core/src/utils/row_addr_remap.rs +++ b/rust/lance-core/src/utils/row_addr_remap.rs @@ -283,6 +283,64 @@ impl CompactRowAddrRemap { } } +// --- rerun fork additions, not present upstream --- +// +// `FragReuseIndex` stores these and is `Debug` + `DeepSizeOf`; upstream does not need +// either yet because it still holds `Vec>>` there. + +impl std::fmt::Debug for RowAddrRemap { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Compact(compact) => f + .debug_struct("RowAddrRemap::Compact") + .field("groups", &compact.num_groups()) + .field("fragments", &compact.num_fragments()) + .finish(), + Self::Direct(map) => f + .debug_struct("RowAddrRemap::Direct") + .field("entries", &map.len()) + .finish(), + } + } +} + +impl CompactRowAddrRemap { + /// Number of rewrite groups held. + pub fn num_groups(&self) -> usize { + self.groups.len() + } + + /// Number of old fragments covered. This is what the structure scales with. + pub fn num_fragments(&self) -> usize { + self.frag_to_group.len() + } +} + +impl crate::deepsize::DeepSizeOf for RowAddrRemap { + fn deep_size_of_children(&self, cx: &mut crate::deepsize::Context) -> usize { + match self { + // Bitmaps dominate; roaring is not `DeepSizeOf`, so approximate from its + // own serialized size rather than under-report it as zero. + Self::Compact(compact) => { + compact + .groups + .iter() + .map(|group| { + group + .frags + .values() + .map(|(bitmap, _)| bitmap.serialized_size() + 16) + .sum::() + + group.new_frag_row_ranges.len() * 16 + }) + .sum::() + + compact.frag_to_group.len() * 12 + } + Self::Direct(map) => map.deep_size_of_children(cx), + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/rust/lance-index/src/vector/bq/storage.rs b/rust/lance-index/src/vector/bq/storage.rs index 72fa8d4b056..4e4bfbfadfd 100644 --- a/rust/lance-index/src/vector/bq/storage.rs +++ b/rust/lance-index/src/vector/bq/storage.rs @@ -2396,7 +2396,7 @@ fn build_frag_reuse_mapping( row_ids: &UInt64Array, ) -> Option>> { let fri = fri?; - if fri.row_id_maps.is_empty() { + if fri.row_addr_maps.is_empty() { return None; } let mut mapping: HashMap> = HashMap::new(); diff --git a/rust/lance-table/src/system_index/frag_reuse.rs b/rust/lance-table/src/system_index/frag_reuse.rs index 40bbc4f58b6..1919d192585 100644 --- a/rust/lance-table/src/system_index/frag_reuse.rs +++ b/rust/lance-table/src/system_index/frag_reuse.rs @@ -7,6 +7,7 @@ use arrow_array::cast::AsArray; use arrow_array::types::UInt64Type; use arrow_array::{Array, ArrayRef, PrimitiveArray, RecordBatch, UInt64Array}; use lance_core::deepsize::{Context, DeepSizeOf}; +use lance_core::utils::row_addr_remap::RowAddrRemap; use lance_core::{Error, Result}; use lance_select::RowAddrTreeMap; use roaring::{RoaringBitmap, RoaringTreemap}; @@ -199,39 +200,60 @@ impl FragReuseIndexDetails { /// An index that stores row ID maps. /// A row ID map describes the mapping from old row address to new address after compactions. /// Each version contains the mapping for one round of compaction. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone)] pub struct FragReuseIndex { pub uuid: Uuid, - pub row_id_maps: Vec>>, + /// One remap per reuse version, oldest first. Order is load-bearing: each version is + /// applied to the previous version's output. + /// + /// Built as [`RowAddrRemap::Compact`] when the index is opened, which costs + /// O(#fragments) rather than the O(#rows) a materialized map would. On one production + /// payload the map form was 676M entries and 40 GB resident, which OOMs the pod. + pub row_addr_maps: Vec, pub details: FragReuseIndexDetails, } impl DeepSizeOf for FragReuseIndex { fn deep_size_of_children(&self, cx: &mut Context) -> usize { - self.row_id_maps.deep_size_of_children(cx) + self.details.deep_size_of_children(cx) + self.row_addr_maps.deep_size_of_children(cx) + self.details.deep_size_of_children(cx) } } impl FragReuseIndex { + /// Build from already-materialized maps, one per version. + /// + /// Kept for callers that hold maps already; it stores them as + /// [`RowAddrRemap::Direct`] and so costs O(#rows). Prefer [`Self::new_from_remaps`]. pub fn new( uuid: Uuid, row_id_maps: Vec>>, details: FragReuseIndexDetails, + ) -> Self { + Self::new_from_remaps( + uuid, + row_id_maps.into_iter().map(RowAddrRemap::direct).collect(), + details, + ) + } + + pub fn new_from_remaps( + uuid: Uuid, + row_addr_maps: Vec, + details: FragReuseIndexDetails, ) -> Self { Self { uuid, - row_id_maps, + row_addr_maps, details, } } pub fn remap_row_id(&self, row_id: u64) -> Option { let mut mapped_value = Some(row_id); - for row_id_map in self.row_id_maps.iter() { + for row_addr_map in self.row_addr_maps.iter() { if mapped_value.is_some() { - mapped_value = row_id_map - .get(&mapped_value.unwrap()) - .copied() + mapped_value = row_addr_map + .get(mapped_value.unwrap()) .unwrap_or(mapped_value); } } diff --git a/rust/lance/src/dataset/optimize/remapping.rs b/rust/lance/src/dataset/optimize/remapping.rs index aef2cd231fc..1a9da714eba 100644 --- a/rust/lance/src/dataset/optimize/remapping.rs +++ b/rust/lance/src/dataset/optimize/remapping.rs @@ -213,7 +213,7 @@ async fn remap_index(dataset: &mut Dataset, index_id: &Uuid) -> Result<()> { .await .unwrap(); - if frag_reuse_index.row_id_maps.is_empty() { + if frag_reuse_index.row_addr_maps.is_empty() { return Ok(()); } @@ -306,10 +306,23 @@ async fn remap_index(dataset: &mut Dataset, index_id: &Uuid) -> Result<()> { // stale (an empty map makes `index::remap_index` return `Keep`). The map is // bounded by the rows the reuse index touched; addresses this index does not // store are simply never looked up. + // + // The compact remap deliberately cannot enumerate its keys: it stores per-fragment + // bitmaps, not rows, and treats any unlisted offset in a rewritten fragment as + // deleted, so "every key" is not a finite set it knows. Rebuild the per-row maps from + // the details here to get the key set. That is O(rows) again, but only on this path, + // which no production caller reaches; the cached open path is what mattered. let composed_row_id_map: HashMap> = frag_reuse_index - .row_id_maps + .details + .versions .iter() - .flat_map(|row_id_map| row_id_map.keys().copied()) + .flat_map(|version| version.groups.iter()) + .flat_map(|group| { + let changed = + RoaringTreemap::deserialize_from(std::io::Cursor::new(&group.changed_row_addrs)) + .expect("fragment reuse index details were already parsed"); + transpose_row_ids_from_digest(changed, &group.old_frags, &group.new_frags).into_keys() + }) .map(|old_addr| (old_addr, frag_reuse_index.remap_row_id(old_addr))) .collect(); diff --git a/rust/lance/src/index/frag_reuse.rs b/rust/lance/src/index/frag_reuse.rs index 23a8fec5145..3ff49e7d01a 100644 --- a/rust/lance/src/index/frag_reuse.rs +++ b/rust/lance/src/index/frag_reuse.rs @@ -2,9 +2,9 @@ // SPDX-FileCopyrightText: Copyright The Lance Authors use crate::Dataset; -use crate::dataset::optimize::remapping::transpose_row_ids_from_digest; use crate::index::DatasetIndexExt; use lance_core::Error; +use lance_core::utils::row_addr_remap::{GroupInput, RowAddrRemap}; use lance_index::frag_reuse::{ FRAG_REUSE_DETAILS_FILE_NAME, FRAG_REUSE_INDEX_NAME, FragReuseGroup, FragReuseIndex, FragReuseIndexDetails, FragReuseVersion, @@ -14,7 +14,6 @@ use lance_table::format::pb::fragment_reuse_index_details::{Content, InlineConte use lance_table::format::pb::{ExternalFile, FragmentReuseIndexDetails}; use prost::Message; use roaring::{RoaringBitmap, RoaringTreemap}; -use std::collections::HashMap; use std::io::Cursor; use std::sync::Arc; use tokio::io::AsyncWriteExt; @@ -72,24 +71,33 @@ pub(crate) async fn open_frag_reuse_index( uuid: Uuid, details: &FragReuseIndexDetails, ) -> lance_core::Result { - let mut row_id_maps: Vec>> = - Vec::with_capacity(details.versions.len()); + // Build the compact form rather than a materialized per-row map. This runs on every + // index open and the result is cached, so the map's O(#rows) cost is paid by readers: + // one production payload here is 88 MB on disk and 40 GB once expanded. + let mut row_addr_maps: Vec = Vec::with_capacity(details.versions.len()); for version in &details.versions { - let mut row_id_map = HashMap::>::new(); + let mut groups = Vec::with_capacity(version.groups.len()); for group in version.groups.iter() { let cursor = Cursor::new(&group.changed_row_addrs); - let changed_row_addrs = RoaringTreemap::deserialize_from(cursor).unwrap(); - let group_row_id_map = transpose_row_ids_from_digest( - changed_row_addrs, - &group.old_frags, - &group.new_frags, - ); - row_id_map.extend(group_row_id_map); + let rewritten_old_row_addrs = RoaringTreemap::deserialize_from(cursor)?; + groups.push(GroupInput { + rewritten_old_row_addrs, + old_frag_ids: group.old_frags.iter().map(|frag| frag.id as u32).collect(), + new_frags: group + .new_frags + .iter() + .map(|frag| (frag.id as u32, frag.physical_rows as u32)) + .collect(), + }); } - row_id_maps.push(row_id_map); + row_addr_maps.push(RowAddrRemap::compact(groups)?); } - Ok(FragReuseIndex::new(uuid, row_id_maps, details.clone())) + Ok(FragReuseIndex::new_from_remaps( + uuid, + row_addr_maps, + details.clone(), + )) } pub(crate) async fn build_new_frag_reuse_index( From f44b10b46c0b5c5b3c86025d0ff5403f4ad472fe Mon Sep 17 00:00:00 2001 From: Ryan Mack Date: Tue, 18 Aug 2026 08:43:07 -0400 Subject: [PATCH 2/3] Run-optimize the compaction row-address set before writing it Forward-port of rerun-io/lance#39 to release-9. The fragment-reuse payload records which row addresses a compaction moved, as a serialized RoaringTreemap. Compaction rewrites long contiguous spans, but nothing calls `RoaringTreemap::optimize()`, so those spans are written as dense bitmap containers at one bit per row. On a real payload that is 88 MB for 676M addresses, essentially all bitmap containers: 676M bits is 84.6 MB. This matters more than it looks. `RowAddrRemap::Compact` keeps those same bitmaps resident, so its 88 MB in-memory footprint is the same containers. Run-optimizing shrinks the stored object, the deserialize on every index open, and the compact remap's own memory. No format change: run containers are already produced today by the binary-copy path via `insert_range`. Co-Authored-By: Claude Opus 5 (1M context) --- rust/lance/src/dataset/optimize.rs | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/rust/lance/src/dataset/optimize.rs b/rust/lance/src/dataset/optimize.rs index 345d10f71db..5cfa3559ead 100644 --- a/rust/lance/src/dataset/optimize.rs +++ b/rust/lance/src/dataset/optimize.rs @@ -1780,7 +1780,13 @@ async fn rewrite_files( let captured_ids = row_ids_rx .try_recv() .map_err(|err| Error::internal(format!("Failed to receive row ids: {}", err)))?; - let row_addrs = captured_ids.row_addrs(None).into_owned(); + let mut row_addrs = captured_ids.row_addrs(None).into_owned(); + // Compaction rewrites long contiguous spans of addresses, which roaring + // stores as dense bitmap containers unless asked otherwise: one bit per row, + // so a large table costs tens of megabytes here. `optimize` converts those to + // run containers, a few bytes per span. This payload is read on every index + // open, so it is worth shrinking at rest. + row_addrs.optimize(); let mut serialized = Vec::with_capacity(row_addrs.serialized_size()); row_addrs.serialize_into(&mut serialized)?; Ok(Some(serialized)) From f11935e1962c23d8d1cda35fd5a90f89e617058f Mon Sep 17 00:00:00 2001 From: Ryan Mack Date: Tue, 18 Aug 2026 09:03:17 -0400 Subject: [PATCH 3/3] Bump h2 and rkyv to clear the cargo-deny advisories Lockfile only, no manifest change; both are patch bumps within the same minor. h2 0.4.15 -> 0.4.16 RUSTSEC-2026-0258 rkyv 0.8.16 -> 0.8.18 RUSTSEC-2026-0233, -0234, -0235 Pre-existing on the branch and unrelated to the change this PR carries. The last green run here was 2026-07-29, before any of the four reached the advisory database: the rkyv entries landed 2026-08-04 and the h2 one 2026-08-18, which is why identical inputs started failing. Neither is urgent on its own merits. rkyv is in the lockfile but not the resolved graph (`cargo tree -i rkyv` matches nothing); the only thing that pulls it is `lindera-dictionary`, behind the optional `tokenizer-lindera` feature, so we never compile it. The h2 advisory is a denial of service driven by a peer sending empty DATA frames, and we are an HTTP/2 client to AWS, so the peer is S3 or DynamoDB behind TLS. Bumping anyway because the fix is free and leaving CI red hides the next one. `cargo deny check` now reports advisories ok, bans ok, licenses ok, sources ok. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 51 +++++++++++++++++++++++++++++++-------------------- 1 file changed, 31 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2dd7494e2a8..de8fe301bfd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -132,7 +132,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -143,7 +143,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -1469,7 +1469,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2770,7 +2770,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2928,7 +2928,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3473,9 +3473,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.15" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" dependencies = [ "atomic-waker", "bytes", @@ -4164,7 +4164,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5770,7 +5770,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7661,9 +7661,9 @@ dependencies = [ [[package]] name = "rkyv" -version = "0.8.16" +version = "0.8.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "73389e0c99e664f919275ab5b5b0471391fe9a8de61e1dff9b1eaf56a90f16e3" +checksum = "d9776093b7ca170454ab1406954f7b7d97a57c51dc6c0642957fb2ef25c2d399" dependencies = [ "bytecheck", "bytes", @@ -7680,13 +7680,13 @@ dependencies = [ [[package]] name = "rkyv_derive" -version = "0.8.16" +version = "0.8.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5d2ed0b54125315fb36bd021e82d314d1c126548f871634b483f46b31d13cac6" +checksum = "1c25ef604ac7dd839d44d64648952ea23c97866f124ff671b0ed2cf3ad9bb06e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] @@ -7817,7 +7817,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7876,7 +7876,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -8422,7 +8422,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -8693,6 +8693,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -8767,10 +8778,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -9794,7 +9805,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]]