Skip to content

fix(core): bound flushed generations folded per merge pass #238

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
beinan merged 1 commit into from
Aug 17, 2026
Merged

fix(core): bound flushed generations folded per merge pass #238

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions crates/lance-context-core/src/datagen_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,14 @@ pub struct DatagenStoreOptions {
/// Merge this writer's flushed generations into the base table after the
/// threshold is reached. `None` or zero disables count-triggered merging.
pub merge_after_generations: Option<usize>,
/// Maximum flushed generations folded into the base table by one merge
/// pass. `None` uses the crate default (8); `Some(0)` means unbounded.
///
/// A merge buffers every row of every generation it takes before appending,
/// so this caps peak merge memory. Leftover generations stay pending for the
/// next pass. Raise it only if merge commits are the bottleneck and the
/// rows are known to be small.
pub merge_max_generations: Option<usize>,
/// Periodically merge this writer's pending generations. `None` or zero
/// disables the timer.
pub cleanup_interval_secs: Option<u64>,
Expand Down Expand Up @@ -92,6 +100,7 @@ impl DatagenStore {
storage_options: options.storage_options.clone(),
shard_id: options.shard_id.clone(),
merge_after_generations: options.merge_after_generations,
merge_max_generations: options.merge_max_generations,
session: None,
schema: Arc::new(datagen_log_schema()),
// Datagen keys on `event_id`, not `id`: event ids are derived
Expand Down
9 changes: 9 additions & 0 deletions crates/lance-context-core/src/generic_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,14 @@ pub struct GenericStoreOptions {
/// Fold this instance's flushed generations into the base table once it has
/// accumulated this many. `None`/`0` disables the count trigger.
pub merge_after_generations: Option<usize>,
/// Maximum flushed generations folded into the base table by one merge
/// pass. `None` uses the crate default (8); `Some(0)` means unbounded.
///
/// A merge buffers every row of every generation it takes before appending,
/// so this caps peak merge memory. Leftover generations stay pending for the
/// next pass. Raise it only if merge commits are the bottleneck and the
/// rows are known to be small.
pub merge_max_generations: Option<usize>,
/// Shared, capacity-bounded Lance session.
pub session: Option<Arc<Session>>,
/// Whether [`GenericStore::add`] seals before returning, making the rows it
Expand Down Expand Up @@ -161,6 +169,7 @@ impl GenericStore {
storage_options: options.storage_options,
shard_id: options.shard_id,
merge_after_generations: options.merge_after_generations,
merge_max_generations: options.merge_max_generations,
session: options.session,
schema: create_schema,
// Always `id`: the LSM merge key, which `SchemaSpec::validate`
Expand Down
29 changes: 29 additions & 0 deletions crates/lance-context-core/src/rollout_store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,6 +337,14 @@ pub struct RolloutStoreOptions {
/// `None` or `0` disables self-merge (the 0.6.0 behavior: generations
/// accumulate and are unioned at read time).
pub merge_after_generations: Option<usize>,
/// Maximum flushed generations folded into the base table by one merge
/// pass. `None` uses the crate default (8); `Some(0)` means unbounded.
///
/// A merge buffers every row of every generation it takes before appending,
/// so this caps peak merge memory. Leftover generations stay pending for the
/// next pass. Raise it only if merge commits are the bottleneck and the
/// rows are known to be small.
pub merge_max_generations: Option<usize>,
/// Shared Lance [`Session`] used to open this store's base dataset (and,
/// transitively, every flushed MemWAL generation it reads — those inherit
/// the base dataset's session).
Expand Down Expand Up @@ -421,6 +429,7 @@ impl RolloutStore {
storage_options,
shard_id,
merge_after_generations,
merge_max_generations,
session,
} = options;
let base = StorageBase::open(
Expand All @@ -429,6 +438,7 @@ impl RolloutStore {
storage_options,
shard_id,
merge_after_generations,
merge_max_generations,
session,
schema: Arc::new(rollout_schema()),
key_column: "id".to_string(),
Expand Down Expand Up @@ -2463,6 +2473,7 @@ mod tests {
// Count trigger disabled: cleanup is the only path that can
// make this row visible, exactly as with flush interval 0.
merge_after_generations: Some(0),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -2505,6 +2516,7 @@ mod tests {
session: None,
shard_id: Some("evicted-0".to_string()),
merge_after_generations: None,
merge_max_generations: None,
};

{
Expand Down Expand Up @@ -2552,6 +2564,7 @@ mod tests {
session: None,
shard_id: Some("observe-0".to_string()),
merge_after_generations: None,
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -2601,6 +2614,7 @@ mod tests {
session: None,
shard_id: Some(shard.to_string()),
merge_after_generations: None,
merge_max_generations: None,
};

let instance_a = RolloutStore::open_with_options(&uri, options("rollout-0"))
Expand Down Expand Up @@ -2820,6 +2834,7 @@ mod tests {
RolloutStoreOptions {
shard_id: Some("refresh-writer".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
..Default::default()
},
)
Expand Down Expand Up @@ -2880,6 +2895,7 @@ mod tests {
RolloutStoreOptions {
shard_id: Some("trajectory-test".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
..Default::default()
},
)
Expand Down Expand Up @@ -2978,6 +2994,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None, // no merge → epoch never reclaimed
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3034,6 +3051,7 @@ mod tests {
// Merge on every append → epoch is reclaimed each time, so
// the following append always hits the reopen path.
merge_after_generations: Some(1),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3071,6 +3089,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None,
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3205,6 +3224,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
// Merge every append into base so each forms its own fragment.
merge_after_generations: Some(1),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3289,6 +3309,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3334,6 +3355,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3401,6 +3423,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(3),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3436,6 +3459,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(3),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3492,6 +3516,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(2),
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -3531,6 +3556,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None, // count trigger off
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -4028,6 +4054,7 @@ mod tests {
RolloutStoreOptions {
shard_id: Some("pagination-benchmark".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
..Default::default()
},
)
Expand Down Expand Up @@ -4124,6 +4151,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None, // disabled
merge_max_generations: None,
},
)
.await
Expand Down Expand Up @@ -4154,6 +4182,7 @@ mod tests {
session: None,
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
},
)
.await
Expand Down
11 changes: 11 additions & 0 deletions crates/lance-context-core/src/store.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -283,6 +283,14 @@ pub struct ContextStoreOptions {
/// and every read unions all of them — the read amplification that
/// previously had no bound at all on this store.
pub merge_after_generations: Option<usize>,
/// Maximum flushed generations folded into the base table by one merge
/// pass. `None` uses the crate default (8); `Some(0)` means unbounded.
///
/// A merge buffers every row of every generation it takes before appending,
/// so this caps peak merge memory. Leftover generations stay pending for the
/// next pass. Raise it only if merge commits are the bottleneck and the
/// rows are known to be small.
pub merge_max_generations: Option<usize>,
/// Whether [`ContextStore::add`] seals the memtable before returning, so the
/// rows it wrote are immediately readable.
///
Expand Down Expand Up @@ -312,6 +320,7 @@ impl Default for ContextStoreOptions {
distance_metric: None,
shard_id: None,
merge_after_generations: None,
merge_max_generations: None,
// Read-your-write by default; see the field docs.
seal_on_add: true,
}
Expand Down Expand Up @@ -591,6 +600,7 @@ impl ContextStore {
storage_options,
shard_id: options.shard_id.clone(),
merge_after_generations: options.merge_after_generations,
merge_max_generations: options.merge_max_generations,
session: None,
schema: Arc::new(arrow_schema.clone()),
key_column: "id".to_string(),
Expand Down Expand Up @@ -2102,6 +2112,7 @@ impl ContextStore {
distance_metric: Some(self.distance_metric),
shard_id: None,
merge_after_generations: None,
merge_max_generations: None,
// A compactor never appends, so the seal mode is irrelevant to it;
// deferring keeps it from ever emitting a generation.
seal_on_add: false,
Expand Down
65 changes: 64 additions & 1 deletion crates/lance-context-core/src/store_base.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,22 @@ pub(crate) const DEFAULT_MANIFEST_SCAN_BATCH_SIZE: usize = 16;
/// concurrently while collecting observability metrics.
pub(crate) const DEFAULT_OBSERVE_CONCURRENCY: usize = 16;

/// Flushed generations folded into the base table by one merge pass, by default.
///
/// A merge buffers every row of every generation it takes before appending, and
/// rollout rows store `binary_payload` inline, so the pass's peak memory is the
/// total blob volume of the generations it took. Unbounded, a worker merging a
/// backlog materialised several GiB at once; freed to glibc but retained in its
/// arenas, that ratcheted RSS up a step per merge until the pod was OOMKilled.
///
/// 8 is deliberately well under the deployed `ROLLOUT_MERGE_AFTER_GENERATIONS`
/// (50 in the deployment that OOMed), so the cap actually binds there, while
/// staying high enough that a merge still amortises its fixed costs -- manifest
/// CAS, base-table commit, directory deletes -- over a useful number of
/// generations. Leftovers are not dropped: they stay pending and the next pass
/// takes them.
pub(crate) const DEFAULT_MERGE_MAX_GENERATIONS: usize = 8;

/// Execute only the first `max_source_fragments` from a Lance compaction plan.
///
/// Lance's built-in `max_source_fragments` stops before a whole planned task
Expand Down Expand Up @@ -194,6 +210,15 @@ pub(crate) struct StorageBaseOptions {
pub shard_id: Option<String>,
/// Count-triggered self-merge threshold; `None`/`0` disables it.
pub merge_after_generations: Option<usize>,
/// Maximum flushed generations folded into the base table by one merge
/// pass. `None`/`0` means unbounded (every pending generation at once).
///
/// This bounds peak merge memory. `read_flushed_generations` buffers every
/// row of every generation it takes, and rollout rows carry `binary_payload`
/// inline, so an unbounded pass over a backlog materialises the full blob
/// volume at once -- the worker OOM this exists to prevent. Leftover
/// generations stay pending and the next pass takes them.
pub merge_max_generations: Option<usize>,
/// Shared, capacity-bounded Lance session. `None` preserves Lance's
/// per-open default (a fresh 6 GiB index + 1 GiB metadata session *per
/// store*, which is the source of unbounded per-append RSS growth).
Expand Down Expand Up @@ -253,6 +278,7 @@ pub(crate) struct StorageBase {
seal_on_put: bool,
/// Self-merge threshold; `0` disables it.
merge_after_generations: usize,
merge_max_generations: usize,
/// Timestamp of the last successful [`Self::compact`] on this handle.
last_compaction: Option<DateTime<Utc>>,
/// Number of successful compactions performed by this handle.
Expand Down Expand Up @@ -305,6 +331,7 @@ impl StorageBase {
storage_options,
shard_id,
merge_after_generations,
merge_max_generations,
session,
schema,
key_column,
Expand Down Expand Up @@ -333,6 +360,7 @@ impl StorageBase {
storage_options,
shard_id,
merge_after_generations,
merge_max_generations,
session,
schema,
key_column,
Expand All @@ -355,6 +383,7 @@ impl StorageBase {
storage_options,
shard_id,
merge_after_generations,
merge_max_generations,
session,
schema,
key_column,
Expand All @@ -378,6 +407,7 @@ impl StorageBase {
latest_schema,
seal_on_put,
merge_after_generations: merge_after_generations.unwrap_or(0),
merge_max_generations: merge_max_generations.unwrap_or(DEFAULT_MERGE_MAX_GENERATIONS),
last_compaction: None,
total_compactions: 0,
last_compaction_error: None,
Expand Down Expand Up @@ -933,7 +963,40 @@ impl StorageBase {
let mut merged_paths: Vec<String> = Vec::new();
let mut batches: Vec<RecordBatch> = Vec::new();
let merge_schema: Arc<Schema> = Arc::new(self.dataset.schema().into());
for flushed in &manifest.flushed_generations {

// Read at most `merge_max_generations` generations per pass.
//
// Every row of every generation is buffered here and stays resident in
// `PreparedMerge.batches` until the commit appends it, so peak memory
// for one merge is the *total* size of the generations taken. Rollout
// rows carry `binary_payload` inline (blob-v2 offload reads back as
// `None` through the LSM scanner, so it cannot be used here), which
// means multi-MB artifacts are in these batches. Unbounded, one pass
// over a large backlog materialises hundreds of MB to several GiB; on
// glibc that memory is freed logically but retained in the allocator's
// arenas, so worker RSS ratchets up a step per merge and never returns.
//
// Capping the *count* is what bounds it, and it is safe because a merge
// of a subset is already a first-class case: `commit_merge` drains only
// the generation ids it actually merged (a relative, retain-not-in-set
// edit) and deletes only those directories, precisely so a concurrent
// flush is not clobbered. Whatever is left over stays pending and the
// next pass takes it -- the same incremental-progress shape as the
// master-side compaction budget in #229.
//
// Generations are the granularity because the drain removes whole ids;
// splitting one generation's rows across two passes would leave rows
// committed to the base table with the generation still listed. That is
// read-safe (the LSM dedups by key) but would re-read and re-append
// those rows on the next pass, so the budget stops at a generation
// boundary.
let budget = if self.merge_max_generations == 0 {
manifest.flushed_generations.len()
} else {
self.merge_max_generations
};

for flushed in manifest.flushed_generations.iter().take(budget) {
let gen_uri = format!(
"{}/_mem_wal/{}/{}",
base_uri, self.write_shard, flushed.path
Expand Down
Loading
Loading