Skip to content

fix: bound MemWAL merge memory with ROLLOUT_MERGE_MAX_BYTES #242

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
Sep 10, 2026
Merged

fix: bound MemWAL merge memory with ROLLOUT_MERGE_MAX_BYTES #242

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
14 changes: 8 additions & 6 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 @@ -51,13 +51,14 @@ pub struct DatagenStoreOptions {
/// 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.
/// pass. `None` uses the crate default (8); `Some(0)` disables this cap.
/// The byte budget applies independently. Leftovers stay pending.
pub merge_max_generations: Option<usize>,
/// Buffered Arrow array byte budget per merge pass. `None` uses 1 GiB;
/// `Some(0)` disables only this cap. Stops after the generation that
/// reaches the budget, or the generation-count cap, whichever comes first.
/// A generation is indivisible, so even an oversized one is fully merged.
pub merge_max_bytes: 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 @@ -101,6 +102,7 @@ impl DatagenStore {
shard_id: options.shard_id.clone(),
merge_after_generations: options.merge_after_generations,
merge_max_generations: options.merge_max_generations,
merge_max_bytes: options.merge_max_bytes,
session: None,
schema: Arc::new(datagen_log_schema()),
// Datagen keys on `event_id`, not `id`: event ids are derived
Expand Down
14 changes: 8 additions & 6 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 @@ -61,13 +61,14 @@ pub struct GenericStoreOptions {
/// 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.
/// pass. `None` uses the crate default (8); `Some(0)` disables this cap.
/// The byte budget applies independently. Leftovers stay pending.
pub merge_max_generations: Option<usize>,
/// Buffered Arrow array byte budget per merge pass. `None` uses 1 GiB;
/// `Some(0)` disables only this cap. Stops after the generation that
/// reaches the budget, or the generation-count cap, whichever comes first.
/// A generation is indivisible, so even an oversized one is fully merged.
pub merge_max_bytes: 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 @@ -170,6 +171,7 @@ impl GenericStore {
shard_id: options.shard_id,
merge_after_generations: options.merge_after_generations,
merge_max_generations: options.merge_max_generations,
merge_max_bytes: options.merge_max_bytes,
session: options.session,
schema: create_schema,
// Always `id`: the LSM merge key, which `SchemaSpec::validate`
Expand Down
34 changes: 28 additions & 6 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 @@ -375,13 +375,14 @@ pub struct RolloutStoreOptions {
/// 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.
/// pass. `None` uses the crate default (8); `Some(0)` disables this cap.
/// The byte budget applies independently. Leftovers stay pending.
pub merge_max_generations: Option<usize>,
/// Buffered Arrow array byte budget per merge pass. `None` uses 1 GiB;
/// `Some(0)` disables only this cap. Stops after the generation that
/// reaches the budget, or the generation-count cap, whichever comes first.
/// A generation is indivisible, so even an oversized one is fully merged.
pub merge_max_bytes: 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 @@ -467,6 +468,7 @@ impl RolloutStore {
shard_id,
merge_after_generations,
merge_max_generations,
merge_max_bytes,
session,
} = options;
let base = StorageBase::open(
Expand All @@ -476,6 +478,7 @@ impl RolloutStore {
shard_id,
merge_after_generations,
merge_max_generations,
merge_max_bytes,
session,
schema: Arc::new(rollout_schema()),
key_column: "id".to_string(),
Expand Down Expand Up @@ -2584,6 +2587,7 @@ mod tests {
// make this row visible, exactly as with flush interval 0.
merge_after_generations: Some(0),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -2627,6 +2631,7 @@ mod tests {
shard_id: Some("evicted-0".to_string()),
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
};

{
Expand Down Expand Up @@ -2675,6 +2680,7 @@ mod tests {
shard_id: Some("observe-0".to_string()),
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -2725,6 +2731,7 @@ mod tests {
shard_id: Some(shard.to_string()),
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
};

let instance_a = RolloutStore::open_with_options(&uri, options("rollout-0"))
Expand Down Expand Up @@ -2945,6 +2952,7 @@ mod tests {
shard_id: Some("refresh-writer".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
..Default::default()
},
)
Expand Down Expand Up @@ -3006,6 +3014,7 @@ mod tests {
shard_id: Some("trajectory-test".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
..Default::default()
},
)
Expand Down Expand Up @@ -3105,6 +3114,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None, // no merge → epoch never reclaimed
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3162,6 +3172,7 @@ mod tests {
// the following append always hits the reopen path.
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3200,6 +3211,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3335,6 +3347,7 @@ mod tests {
// Merge every append into base so each forms its own fragment.
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3420,6 +3433,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3466,6 +3480,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3534,6 +3549,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(3),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3570,6 +3586,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(3),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3627,6 +3644,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(2),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -3667,6 +3685,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None, // count trigger off
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -4165,6 +4184,7 @@ mod tests {
shard_id: Some("pagination-benchmark".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
..Default::default()
},
)
Expand Down Expand Up @@ -4262,6 +4282,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: None, // disabled
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down Expand Up @@ -4293,6 +4314,7 @@ mod tests {
shard_id: Some("rollout-0".to_string()),
merge_after_generations: Some(1),
merge_max_generations: None,
merge_max_bytes: None,
},
)
.await
Expand Down
16 changes: 10 additions & 6 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 @@ -284,13 +284,14 @@ pub struct ContextStoreOptions {
/// 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.
/// pass. `None` uses the crate default (8); `Some(0)` disables this cap.
/// The byte budget applies independently. Leftovers stay pending.
pub merge_max_generations: Option<usize>,
/// Buffered Arrow array byte budget per merge pass. `None` uses 1 GiB;
/// `Some(0)` disables only this cap. Stops after the generation that
/// reaches the budget, or the generation-count cap, whichever comes first.
/// A generation is indivisible, so even an oversized one is fully merged.
pub merge_max_bytes: Option<usize>,
/// Whether [`ContextStore::add`] seals the memtable before returning, so the
/// rows it wrote are immediately readable.
///
Expand Down Expand Up @@ -321,6 +322,7 @@ impl Default for ContextStoreOptions {
shard_id: None,
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: None,
// Read-your-write by default; see the field docs.
seal_on_add: true,
}
Expand Down Expand Up @@ -601,6 +603,7 @@ impl ContextStore {
shard_id: options.shard_id.clone(),
merge_after_generations: options.merge_after_generations,
merge_max_generations: options.merge_max_generations,
merge_max_bytes: options.merge_max_bytes,
session: None,
schema: Arc::new(arrow_schema.clone()),
key_column: "id".to_string(),
Expand Down Expand Up @@ -2119,6 +2122,7 @@ impl ContextStore {
shard_id: None,
merge_after_generations: None,
merge_max_generations: None,
merge_max_bytes: 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
Loading
Loading