From 1e5fef5809a62d65dbb05371dd5b87680e87a1e3 Mon Sep 17 00:00:00 2001 From: "Lei, HUANG" <6406592+v0y4g3r@users.noreply.github.com> Date: Tue, 20 Oct 2026 08:31:26 -0700 Subject: [PATCH] fix(mito): backport bulk memtable version ordering fix to release/5e02efa9 (#9789) * fix(mito2): preserve last-non-null values across compaction (#9169) * fix(mito2): preserve last-non-null values during compaction Use a dedicated LastNonNullPicker to close TWCS seeds over overlapping SSTs, merge connected seed groups, and defer closures with busy inputs. Keep LastRow and strict-window selection unchanged. Index overlap candidates on the blocking compaction runtime and cover field promotion, transitive dependencies, expiration, and conservative key-range pruning with regression tests. Refs #9146 Signed-off-by: Lei, HUANG * fix(mito2): guard last-non-null compaction against pending memtables Track conservative memtable sequence lower bounds and capture them from the same version as compaction inputs. Defer unsafe SST closures and strict-window plans until pending writes no longer intersect their sequence span. Cache minimum sequences during bulk conversion and expose a lightweight memtable getter to avoid rescanning rows or computing full statistics. Fixes #9157 Signed-off-by: Lei, HUANG * fix(mito2): limit eligible last-non-null compaction outputs Enumerate all TWCS seeds for LastNonNull while retaining bounded window planning concurrency. Apply the output budget after closure expansion and busy/memtable safety checks, preserving execution priority and ordinary TWCS behavior. Cover rejected and absorbed seeds, priority ordering, and bounded and unlimited output counts with real-picker regression tests. Signed-off-by: Lei, HUANG * fix(mito2): include all candidates before last-non-null closure Signed-off-by: Lei, HUANG * refactor(mito2): clarify last-non-null compaction closure Extract transitive closure expansion from output scheduling and clarify seed handling. Document the dedicated picker rationale and schema-evolution risks in PK overlap checks. Signed-off-by: Lei, HUANG * test(mito2): cover deferred non-seed compaction bridges Signed-off-by: Lei, HUANG * fix(mito2): defer strict-window compaction with busy dependencies Keep busy SSTs when building the LastNonNull strict-window dependency closure, then defer the entire rewrite plan if any selected input is busy. Preserve LastRow filtering and expired-file cleanup.\n\nCover requested, transitive and unrelated busy windows, including retry after the dependency is released. Signed-off-by: Lei, HUANG * docs(mito2): document compaction overlap index fields Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG (cherry picked from commit 259b42c03920997d92a16802562f961ad7692da4) Signed-off-by: Lei, HUANG * fix(mito): preserve version ordering in bulk merges and flushes (#9517) * fix(mito): preserve delete barriers in flat last-non-null dedup Signed-off-by: evenyag * fix(mito): respect sequence ranges in bulk merges and flushes Signed-off-by: evenyag * perf(mito): avoid merge selection on the bulk write path Signed-off-by: evenyag --------- Signed-off-by: evenyag (cherry picked from commit 674ba7266949c3c05b71e6e7fb99891a71cc8d2e) Signed-off-by: Lei, HUANG * test(mito2): match release empty-result rendering in bulk flush test Signed-off-by: Lei, HUANG --------- Signed-off-by: Lei, HUANG Signed-off-by: evenyag Co-authored-by: Yingwen --- .../benches/bench_filter_time_partition.rs | 2 + src/mito2/src/access_layer.rs | 2 +- src/mito2/src/compaction.rs | 2 + src/mito2/src/compaction/compactor.rs | 17 +- src/mito2/src/compaction/last_non_null.rs | 610 ++++++++++++++++++ src/mito2/src/compaction/overlap.rs | 293 +++++++++ src/mito2/src/compaction/picker.rs | 25 +- .../src/compaction/scheduler/planning.rs | 12 +- src/mito2/src/compaction/test_util.rs | 45 ++ src/mito2/src/compaction/twcs.rs | 106 +-- src/mito2/src/compaction/window.rs | 100 ++- src/mito2/src/engine/flush_test.rs | 309 ++++++++- src/mito2/src/engine/merge_mode_test.rs | 255 +++++++- src/mito2/src/flush.rs | 330 +++++----- src/mito2/src/memtable.rs | 35 + src/mito2/src/memtable/bulk.rs | 594 +++++++++++------ src/mito2/src/memtable/bulk/part.rs | 98 ++- .../src/memtable/simple_bulk_memtable.rs | 45 +- src/mito2/src/memtable/stats.rs | 3 + src/mito2/src/memtable/time_partition.rs | 3 + src/mito2/src/memtable/time_series.rs | 42 +- src/mito2/src/memtable/version.rs | 11 + src/mito2/src/read/flat_dedup.rs | 163 +++-- src/mito2/src/region_write_ctx.rs | 3 + src/mito2/src/test_util/memtable_util.rs | 4 + src/mito2/src/worker/handle_bulk_insert.rs | 1 + src/mito2/src/worker/handle_write.rs | 1 + 27 files changed, 2579 insertions(+), 532 deletions(-) create mode 100644 src/mito2/src/compaction/last_non_null.rs create mode 100644 src/mito2/src/compaction/overlap.rs diff --git a/src/mito2/benches/bench_filter_time_partition.rs b/src/mito2/benches/bench_filter_time_partition.rs index 0c394aaf8f9..d44f33920a9 100644 --- a/src/mito2/benches/bench_filter_time_partition.rs +++ b/src/mito2/benches/bench_filter_time_partition.rs @@ -57,6 +57,7 @@ fn random_array(num: usize) -> BulkPart { max_timestamp: max, min_timestamp: min, sequence: 0, + min_sequence: 0, timestamp_index: 0, raw_data: None, } @@ -90,6 +91,7 @@ fn filter_arrow_impl(part: &BulkPart, min: i64, max: i64) -> Option { max_timestamp: max, min_timestamp: min, sequence: 0, + min_sequence: 0, timestamp_index: part.timestamp_index, raw_data: None, }) diff --git a/src/mito2/src/access_layer.rs b/src/mito2/src/access_layer.rs index 58b24c3c350..90f02f8ee32 100644 --- a/src/mito2/src/access_layer.rs +++ b/src/mito2/src/access_layer.rs @@ -582,7 +582,7 @@ pub struct SstWriteRequest { #[allow(dead_code)] pub storage: Option, /// Optional uniform row sequence for writes that do not preserve sequences. - /// Compaction passes `None` to retain the reader's effective input sequences. + /// Compaction and LastNonNull flushes pass `None` to retain input sequences. pub max_sequence: Option, pub sst_write_format: FormatType, diff --git a/src/mito2/src/compaction.rs b/src/mito2/src/compaction.rs index e7f74b22648..1fbbfad5613 100644 --- a/src/mito2/src/compaction.rs +++ b/src/mito2/src/compaction.rs @@ -15,7 +15,9 @@ mod buckets; pub mod compactor; mod json2; +mod last_non_null; pub mod memory_manager; +mod overlap; pub mod picker; mod reader; pub mod run; diff --git a/src/mito2/src/compaction/compactor.rs b/src/mito2/src/compaction/compactor.rs index 598c0b7f207..26eb9108565 100644 --- a/src/mito2/src/compaction/compactor.rs +++ b/src/mito2/src/compaction/compactor.rs @@ -30,7 +30,7 @@ use snafu::{OptionExt, ResultExt}; use store_api::ManifestVersion; use store_api::metadata::RegionMetadataRef; use store_api::region_request::PathType; -use store_api::storage::RegionId; +use store_api::storage::{RegionId, SequenceNumber}; use crate::access_layer::{ AccessLayer, AccessLayerRef, Metrics, OperationType, SstWriteRequest, WriteType, @@ -47,7 +47,7 @@ use crate::error::{ }; use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList}; use crate::manifest::manager::{RegionManifestManager, RegionManifestOptions}; -use crate::region::options::RegionOptions; +use crate::region::options::{MergeMode, RegionOptions}; use crate::region::version::VersionRef; use crate::region::{ManifestContext, RegionLeaderState, RegionRoleState}; use crate::schedule::scheduler::LocalScheduler; @@ -73,6 +73,10 @@ pub struct CompactionVersion { pub(crate) options: RegionOptions, /// SSTs of the region. pub(crate) ssts: SstVersionRef, + /// Lower bound of pending memtable sequences from the same Version as `ssts`. + /// None means no barrier; Some(0) conservatively represents an unknown bound. + /// Keep only the value so background compaction cannot pin flushed memtables. + pub(crate) memtable_min_sequence: Option, /// Inferred compaction time window. pub(crate) compaction_time_window: Option, } @@ -83,6 +87,11 @@ impl From for CompactionVersion { metadata: value.metadata.clone(), options: value.options.clone(), ssts: value.ssts.clone(), + memtable_min_sequence: if value.options.merge_mode() == MergeMode::LastNonNull { + value.memtables.min_sequence() + } else { + None + }, compaction_time_window: value.compaction_time_window, } } @@ -205,6 +214,9 @@ pub async fn open_compaction_region( metadata: region_metadata.clone(), options: req.region_options.clone(), ssts: Arc::new(ssts), + // Remote execution uses an already picked output. A manifest-only + // opener cannot prove that a new LastNonNull pick is memtable-safe. + memtable_min_sequence: Some(0), compaction_time_window: manifest.compaction_time_window, } }; @@ -1160,6 +1172,7 @@ mod tests { metadata, options: RegionOptions::default(), ssts: Arc::new(SstVersion::new()), + memtable_min_sequence: None, compaction_time_window: None, }, file_purger: None, diff --git a/src/mito2/src/compaction/last_non_null.rs b/src/mito2/src/compaction/last_non_null.rs new file mode 100644 index 00000000000..60f9bc254af --- /dev/null +++ b/src/mito2/src/compaction/last_non_null.rs @@ -0,0 +1,610 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::{HashMap, HashSet}; + +use snafu::ResultExt; +use store_api::metadata::RegionMetadata; +use store_api::storage::SequenceNumber; + +use crate::compaction::CompactionOutput; +use crate::compaction::compactor::CompactionRegion; +use crate::compaction::overlap::FileOverlapIndex; +use crate::compaction::picker::{Picker, PickerOutput}; +use crate::compaction::twcs::TwcsPicker; +use crate::error::{JoinSnafu, Result}; +use crate::sst::file::{FileHandle, RegionFileId}; + +/// Picker for [`MergeMode::LastNonNull`](crate::region::options::MergeMode). +/// +/// LastNonNull fills null fields of a newer row from older non-null values when +/// rows with the same key and timestamp are merged. That fill is only correct if +/// every SST that can contribute to the same logical row is merged together: +/// once a value is promoted into an output, an intermediate version left behind +/// in a separately compacted SST is invisible, so it can no longer override the +/// promoted value on a later read. +/// +/// TWCS alone cannot guarantee this: it schedules independent windows and seeds, +/// while an L1 or cross-window SST can overlap several of them. This picker +/// therefore delegates seed selection to [TwcsPicker] and then closes each seed +/// over the whole snapshot: it absorbs every overlapping file and every seed +/// connected through those overlaps, so a merge stream is never split across +/// outputs, and defers a closure whose inputs are already compacting. +#[derive(Debug)] +pub(super) struct LastNonNullPicker { + base_picker: TwcsPicker, +} + +impl LastNonNullPicker { + pub(super) fn new(base_picker: TwcsPicker) -> Self { + Self { base_picker } + } +} + +#[async_trait::async_trait] +impl Picker for LastNonNullPicker { + async fn pick(&self, region: &CompactionRegion) -> Result> { + let max_outputs = self.base_picker.max_background_tasks; + + // Find all possible compaction seeds + let Some(mut all_seeds) = self + .base_picker + .pick_with_output_limit(region, None) + .await? + else { + return Ok(None); + }; + let version = region.current_version.clone(); + common_runtime::spawn_blocking_compact(move || { + let expired: HashSet<_> = all_seeds + .expired_ssts + .iter() + .map(FileHandle::file_id) + .collect(); + // Include busy files and all levels/windows. Request ranges and + // TWCS size/file-count limits only constrain seeds, not correctness. + let all_files = version + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .filter(|file| !expired.contains(&file.file_id())) + .cloned() + .collect(); + all_seeds.outputs = + build_closed_outputs(all_seeds.outputs, all_files, &version.metadata); + all_seeds.outputs.retain(|output| { + inputs_precede_memtables(&output.inputs, version.memtable_min_sequence) + }); + if let Some(limit) = max_outputs { + // Only eligible closures consume the budget. Keep the high-priority + // tail because the compactor pops from the end of the output list. + let excess = all_seeds.outputs.len().saturating_sub(limit); + all_seeds.outputs.drain(..excess); + } + (!all_seeds.outputs.is_empty() || !all_seeds.expired_ssts.is_empty()) + .then_some(all_seeds) + }) + .await + .context(JoinSnafu) + } +} + +/// Imported SSTs can be newer than pending writes. A merge must not attach an +/// old field to a sequence above an intermediate version left in a memtable. +pub(super) fn inputs_precede_memtables( + inputs: &[FileHandle], + memtable_min_sequence: Option, +) -> bool { + memtable_min_sequence.is_none_or(|min_sequence| { + inputs.iter().all(|file| { + file.meta_ref() + .sequence + .is_some_and(|max_sequence| max_sequence.get() < min_sequence) + }) + }) +} + +/// Each seed and snapshot file is consumed at most once. Absorbing a seed also +/// absorbs its disconnected members: a merge stream must never be split across +/// independently scheduled outputs. +fn build_closed_outputs( + seeds: Vec, + files: Vec, + metadata: &RegionMetadata, +) -> Vec { + // Maps every seed input to its seed, so a closure that reaches that input + // can absorb the rest of the seed instead of letting it schedule separately. + let seed_by_file: HashMap<_, _> = seeds + .iter() + .enumerate() + .flat_map(|(i, seed)| seed.inputs.iter().map(move |file| (file.file_id(), i))) + .collect(); + // `None` marks a seed already absorbed by an earlier closure. + let mut seeds: Vec<_> = seeds.into_iter().map(Some).collect(); + let mut remaining = FileOverlapIndex::new(files, metadata); + let mut outputs = Vec::new(); + // The compactor pops from the end; retain TWCS's seed priority. + for seed_index in (0..seeds.len()).rev() { + let Some(output) = seeds[seed_index].take() else { + continue; + }; + let output = expand_closure(output, &mut seeds, &seed_by_file, &mut remaining); + // Defer the entire closure rather than truncate it around busy files. + if !output.inputs.iter().any(FileHandle::compacting) { + outputs.push(output); + } + } + outputs.reverse(); + outputs +} + +/// Expands one seed into its full merge closure: absorb every connected seed +/// and every overlapping snapshot file, transitively. Growing `output.inputs` +/// while iterating is what makes the expansion transitive. +fn expand_closure( + mut output: CompactionOutput, + seeds: &mut [Option], + seed_by_file: &HashMap, + remaining_files: &mut FileOverlapIndex<'_>, +) -> CompactionOutput { + let mut selected = HashSet::new(); + let inputs = std::mem::take(&mut output.inputs); + extend_inputs(&mut output, inputs, &mut selected, remaining_files); + + let mut cursor = 0; + while cursor < output.inputs.len() { + let input = output.inputs[cursor].clone(); + if let Some(&seed_index) = seed_by_file.get(&input.file_id()) + && let Some(seed) = seeds[seed_index].take() + { + // AND the tombstone policies so the merged stream stays conservative. + output.filter_deleted &= seed.filter_deleted; + extend_inputs(&mut output, seed.inputs, &mut selected, remaining_files); + } + let dependencies = remaining_files.drain_overlaps(&input); + extend_inputs(&mut output, dependencies, &mut selected, remaining_files); + cursor += 1; + } + output +} + +/// Appends never-seen inputs and removes them from the overlap index so the +/// same file is neither expanded nor scheduled twice. +fn extend_inputs( + output: &mut CompactionOutput, + inputs: Vec, + selected: &mut HashSet, + remaining: &mut FileOverlapIndex<'_>, +) { + for input in inputs { + if selected.insert(input.file_id()) { + remaining.remove(input.file_id()); + output.inputs.push(input); + } + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use api::v1::region::compact_request; + use common_time::Timestamp; + use common_time::range::TimestampRange; + use store_api::storage::FileId; + + use super::*; + use crate::compaction::picker::new_picker; + use crate::compaction::test_util::{ + compaction_region_with_ssts, new_file_handle, new_file_handle_with_size_and_sequence, + }; + use crate::region::options::{CompactionOptions, MergeMode}; + use crate::sst::file::{FileMeta, RegionFileId}; + use crate::test_util::memtable_util::metadata_for_test; + + fn seed(inputs: Vec) -> CompactionOutput { + CompactionOutput { + output_level: 1, + inputs, + filter_deleted: false, + output_time_range: None, + } + } + + fn file_ids(files: &[FileHandle]) -> HashSet { + files.iter().map(FileHandle::file_id).collect() + } + + #[rstest::rstest] + #[case::memtable_barrier(2, Some(20), false, Some(1), vec![vec![1]])] + #[case::busy_dependency(2, None, true, Some(1), vec![vec![1]])] + #[case::highest_priority(2, None, false, Some(1), vec![vec![2]])] + #[case::priority_order(2, None, false, Some(2), vec![vec![2], vec![1]])] + #[case::unlimited(2, None, false, None, vec![vec![2], vec![1], vec![0]])] + #[case::fewer_eligible_outputs(2, Some(20), false, Some(3), vec![vec![1], vec![0]])] + #[case::absorbed_seeds(1, None, false, Some(2), vec![vec![2, 1], vec![0]])] + #[case::busy_bridge_defers_both_seeds(1, None, true, Some(2), vec![vec![0]])] + #[case::memtable_barrier_defers_both_seeds(1, Some(20), false, Some(2), vec![vec![0]])] + #[tokio::test] + async fn test_output_limit_counts_eligible_closures( + #[case] dependency_start_window: usize, + #[case] memtable_min: Option, + #[case] busy: bool, + #[case] limit: Option, + #[case] expected_windows: Vec>, + ) { + // Three independent, eligible L0 windows. A large L1 dependency is not + // a seed, but can reject the newest closure or connect two seeds. + // Rejecting that bridge must defer both connected seeds, leaving only + // the independent oldest window eligible rather than a partial merge. + let windows: Vec> = (0..3) + .map(|window| { + (0..4) + .map(|i| FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_second(window * 3600), + Timestamp::new_second(window * 3600 + 10), + ), + level: 0, + file_size: 100, + sequence: std::num::NonZeroU64::new((window * 4 + i + 1) as u64), + ..Default::default() + }) + .collect() + }) + .collect(); + let dependency = FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_second(dependency_start_window as i64 * 3600), + Timestamp::new_second(7210), + ), + level: 1, + file_size: 1_000_000, + sequence: std::num::NonZeroU64::new(20), + ..Default::default() + }; + let dependency_id = dependency.file_id(); + let files = windows.iter().flatten().cloned().chain([dependency]); + let mut region = compaction_region_with_ssts(files, Duration::from_secs(60)).await; + region.ttl = None; + region.region_options.merge_mode = Some(MergeMode::LastNonNull); + region.current_version.memtable_min_sequence = memtable_min; + let CompactionOptions::Twcs(opts) = &mut region.region_options.compaction; + opts.time_window = Some(Duration::from_secs(3600)); + opts.active_window_trigger_file_num = 4; + opts.inactive_window_trigger_file_num = 4; + region.current_version.ssts.levels()[1] + .files() + .next() + .unwrap() + .set_compacting(busy); + let request = compact_request::Options::Regular(Default::default()); + let picker = new_picker(&request, ®ion.region_options, limit, None); + + let picked = picker.pick(®ion).await.unwrap().unwrap(); + assert_eq!(expected_windows.len(), picked.outputs.len()); + assert!(picked.expired_ssts.is_empty()); + // The execution pops outputs from the end, not the front. + for (output, window_indices) in picked.outputs.iter().rev().zip(expected_windows) { + let mut expected: HashSet<_> = window_indices + .iter() + .flat_map(|&i| windows[i].iter().map(FileMeta::file_id)) + .collect(); + if window_indices.contains(&2) { + expected.insert(dependency_id); + } + assert_eq!(expected, file_ids(&output.inputs)); + } + + // The LastRow picker must retain its original seed cap and priority. + region.region_options.merge_mode = Some(MergeMode::LastRow); + let picker = new_picker(&request, ®ion.region_options, limit, None); + let picked = picker.pick(®ion).await.unwrap().unwrap(); + assert_eq!(limit.unwrap_or(3).min(3), picked.outputs.len()); + for (output, window) in picked.outputs.iter().rev().zip(windows.iter().rev()) { + let expected: HashSet<_> = window.iter().map(FileMeta::file_id).collect(); + assert_eq!(expected, file_ids(&output.inputs)); + } + } + + // Cover absent/unknown boundaries and strict inequality without arithmetic + // on sequence numbers, including the maximum representable sequence. + #[rstest::rstest] + #[case(0, None, true)] + #[case(0, Some(2), false)] + #[case(1, Some(0), false)] + #[case(1, Some(2), true)] + #[case(2, Some(2), false)] + #[case(3, Some(2), false)] + #[case(u64::MAX - 1, Some(u64::MAX), true)] + #[case(u64::MAX, Some(u64::MAX), false)] + fn test_inputs_must_precede_pending_memtables( + #[case] file_sequence: u64, + #[case] memtable_min: Option, + #[case] expected: bool, + ) { + let file = + new_file_handle_with_size_and_sequence(FileId::random(), 0, 10, 0, file_sequence, 100); + assert_eq!(expected, inputs_precede_memtables(&[file], memtable_min)); + } + + #[test] + fn test_transitive_closure_exceeds_seed_limit_across_levels_and_windows() { + let metadata = metadata_for_test(); + // Inclusive endpoints connect neighbors across hours and both levels. + // A one-pass expansion or applying the TWCS seed cap loses the tail. + let files: Vec<_> = (0..32) + .map(|i| { + new_file_handle( + FileId::random(), + i * 3_600_000, + (i + 1) * 3_600_000, + (i % 2) as u8, + ) + }) + .collect(); + let outputs = + build_closed_outputs(vec![seed(vec![files[0].clone()])], files.clone(), &metadata); + assert_eq!(1, outputs.len()); + assert_eq!(files.len(), outputs[0].inputs.len()); + assert_eq!(file_ids(&files), file_ids(&outputs[0].inputs)); + } + + #[rstest::rstest] + #[case(false)] + #[case(true)] + fn test_absorbed_seed_closes_disconnected_members_and_defers_busy_group(#[case] busy: bool) { + let metadata = metadata_for_test(); + let files: Vec<_> = [(0, 10), (10, 20), (100, 110), (110, 120), (200, 210)] + .into_iter() + .map(|(start, end)| new_file_handle(FileId::random(), start, end, 0)) + .collect(); + files[3].set_compacting(busy); + let mut priority_seed = seed(vec![files[0].clone()]); + priority_seed.filter_deleted = true; + let seeds = vec![ + seed(vec![files[4].clone()]), + seed(vec![files[1].clone(), files[2].clone()]), + priority_seed, + ]; + + let mut outputs = build_closed_outputs(seeds, files.clone(), &metadata); + assert_eq!(if busy { 1 } else { 2 }, outputs.len()); + if !busy { + let first = outputs.pop().unwrap(); + assert_eq!(4, first.inputs.len()); + assert_eq!(file_ids(&files[..4]), file_ids(&first.inputs)); + // Preserve the absorbed seed's conservative tombstone policy. + assert!(!first.filter_deleted); + } + assert_eq!( + vec![files[4].file_id()], + outputs[0] + .inputs + .iter() + .map(FileHandle::file_id) + .collect::>() + ); + } + + #[rstest::rstest] + #[case("chain", 2048)] + #[case("dense_with_external", 1024)] + #[case("time_dense_pk_disjoint", 1)] + fn test_large_snapshot_closure(#[case] shape: &str, #[case] expected_count: usize) { + use bytes::Bytes; + + use crate::compaction::test_util::new_file_handle_with_size_sequence_and_primary_key_range; + + let mut metadata = (*metadata_for_test()).clone(); + metadata.region_id = 0.into(); + let files = (0..2048) + .map(|i| { + let (start, end, pk) = match shape { + "chain" => (i, i + 1, None), + "dense_with_external" if i < 1024 => (0, 1, None), + "dense_with_external" => (i, i, None), + "time_dense_pk_disjoint" => { + let pk = Bytes::copy_from_slice(&(i as u32).to_be_bytes()); + (0, 1, Some((pk.clone(), pk))) + } + _ => unreachable!(), + }; + new_file_handle_with_size_sequence_and_primary_key_range( + FileId::random(), + start, + end, + 0, + i as u64 + 1, + 100, + pk, + ) + }) + .collect::>(); + let outputs = + build_closed_outputs(vec![seed(vec![files[0].clone()])], files.clone(), &metadata); + assert_eq!(1, outputs.len()); + assert_eq!(expected_count, outputs[0].inputs.len()); + assert_eq!( + file_ids(&files[..expected_count]), + file_ids(&outputs[0].inputs) + ); + } + + #[rstest::rstest] + #[case(MergeMode::LastRow, false, None, 4)] + #[case(MergeMode::LastRow, true, None, 4)] + #[case(MergeMode::LastNonNull, false, None, 5)] + #[case(MergeMode::LastNonNull, true, None, 0)] + #[case(MergeMode::LastRow, false, Some(5), 4)] + #[case(MergeMode::LastNonNull, false, Some(5), 0)] + #[case(MergeMode::LastNonNull, false, Some(6), 5)] + #[tokio::test] + async fn test_picker_dependencies_outside_request_window( + #[case] merge_mode: MergeMode, + #[case] busy: bool, + #[case] memtable_min: Option, + #[case] expected_count: usize, + ) { + let files = (0..5).map(|i| FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_second(0), + Timestamp::new_second(if i == 4 { 3601 } else { 10 }), + ), + level: if i == 4 { 1 } else { 0 }, + file_size: if i == 4 { 1_000_000 } else { 100 }, + sequence: std::num::NonZeroU64::new(i + 1), + ..Default::default() + }); + let mut region = compaction_region_with_ssts(files, Duration::from_secs(60)).await; + region.ttl = None; + region.region_options.merge_mode = Some(merge_mode); + region.current_version.memtable_min_sequence = memtable_min; + let CompactionOptions::Twcs(opts) = &mut region.region_options.compaction; + opts.time_window = Some(Duration::from_secs(3600)); + let dependency = region.current_version.ssts.levels()[1] + .files() + .next() + .unwrap(); + dependency.set_compacting(busy); + let picker = new_picker( + &compact_request::Options::Regular(Default::default()), + ®ion.region_options, + Some(1), + TimestampRange::new(Timestamp::new_second(0), Timestamp::new_second(3600)), + ); + + let picked = picker.pick(®ion).await.unwrap(); + if expected_count == 0 { + assert!(picked.is_none()); + } else { + let picked = picked.unwrap(); + assert_eq!(1, picked.outputs.len()); + assert_eq!(expected_count, picked.outputs[0].inputs.len()); + assert_eq!( + merge_mode == MergeMode::LastNonNull, + picked.outputs[0] + .inputs + .iter() + .any(|file| file.file_id() == dependency.file_id()) + ); + } + } + + #[rstest::rstest] + #[case(MergeMode::LastNonNull, None, 3)] + #[case(MergeMode::LastNonNull, Some(2), 0)] + #[case(MergeMode::LastRow, Some(2), 3)] + #[tokio::test] + async fn test_strict_window_keeps_all_versions_in_disjoint_output_slices( + #[case] merge_mode: MergeMode, + #[case] memtable_min: Option, + #[case] expected_outputs: usize, + ) { + let files: Vec<_> = [(0, 10), (10, 10), (10, 20)] + .into_iter() + .enumerate() + .map(|(i, (start, end))| { + new_file_handle_with_size_and_sequence( + FileId::random(), + start * 1000, + end * 1000, + 0, + i as u64 + 1, + 100, + ) + .meta_ref() + .clone() + }) + .collect(); + let mut region = compaction_region_with_ssts(files, Duration::from_secs(60)).await; + region.region_options.merge_mode = Some(merge_mode); + region.current_version.options.merge_mode = Some(merge_mode); + region.current_version.memtable_min_sequence = memtable_min; + let picker = new_picker( + &compact_request::Options::StrictWindow(api::v1::region::StrictWindow { + window_seconds: 10, + }), + ®ion.region_options, + Some(1), + TimestampRange::new(Timestamp::new_second(10), Timestamp::new_second(20)), + ); + + let picked = picker.pick(®ion).await.unwrap().unwrap(); + // Cross-window inputs also need their outer slices rewritten, despite + // requesting only [10, 20). All three versions at t=10 share one stream. + // With pending sequence 2, [0, 10) alone is safe, but must not be + // rewritten without the unsafe windows sharing its input SST. + assert_eq!(expected_outputs, picked.outputs.len()); + for (i, output) in picked.outputs.iter().enumerate() { + assert_eq!( + TimestampRange::new( + Timestamp::new_second(i as i64 * 10), + Timestamp::new_second((i as i64 + 1) * 10), + ), + output.output_time_range + ); + assert_eq!(if i == 1 { 3 } else { 1 }, output.inputs.len()); + assert!(!output.filter_deleted); + } + } + + #[tokio::test] + async fn test_busy_closure_keeps_expired_files_without_rewriting_them() { + let now = Timestamp::current_millis().value(); + let files = (0..6).map(|i| FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(if i == 5 { 0 } else { now - 10_000 }), + Timestamp::new_millisecond(if i == 5 { 10 } else { now }), + ), + level: if i == 4 { 1 } else { 0 }, + file_size: if i == 4 { 1_000_000 } else { 100 }, + sequence: std::num::NonZeroU64::new(i + 1), + ..Default::default() + }); + let mut region = compaction_region_with_ssts(files, Duration::from_secs(3600)).await; + region.region_options.merge_mode = Some(MergeMode::LastNonNull); + let dependency = region.current_version.ssts.levels()[1] + .files() + .next() + .unwrap(); + dependency.set_compacting(true); + let picker = new_picker( + &compact_request::Options::Regular(Default::default()), + ®ion.region_options, + Some(1), + None, + ); + let picked = picker.pick(®ion).await.unwrap().unwrap(); + assert!(picked.outputs.is_empty()); + assert_eq!(1, picked.expired_ssts.len()); + assert_eq!( + Timestamp::new_millisecond(10), + picked.expired_ssts[0].time_range().1 + ); + assert!(!picked.expired_ssts[0].compacting()); + + dependency.set_compacting(false); + let picked = picker.pick(®ion).await.unwrap().unwrap(); + assert_eq!(1, picked.outputs.len()); + assert_eq!(5, picked.outputs[0].inputs.len()); + assert_eq!(1, picked.expired_ssts.len()); + assert!(!file_ids(&picked.outputs[0].inputs).contains(&picked.expired_ssts[0].file_id())); + } +} diff --git a/src/mito2/src/compaction/overlap.rs b/src/mito2/src/compaction/overlap.rs new file mode 100644 index 00000000000..38cb7ad8a4a --- /dev/null +++ b/src/mito2/src/compaction/overlap.rs @@ -0,0 +1,293 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; +use std::ops::Range; + +use common_time::Timestamp; +use store_api::metadata::RegionMetadata; + +use crate::compaction::run::primary_key_ranges_overlap; +use crate::sst::file::{FileHandle, RegionFileId}; + +/// Snapshot-local overlap candidates, ordered by start time. Each subtree stores +/// its maximum active end. Removing visited files prunes dense internal overlaps. +/// PK-disjoint files with overlapping times can still cost O(N) per query, so a +/// complete closure has O(N²) worst-case time and O(N) auxiliary space. +pub(super) struct FileOverlapIndex<'a> { + /// Current region metadata used to decide whether PK bounds can safely exclude overlaps. + metadata: &'a RegionMetadata, + /// Candidates in original snapshot order, retained after removal so their + /// indices remain stable and drained matches can preserve merge input order. + files: Vec, + /// Indices into `files`, sorted by start time with the snapshot index as a + /// tie-breaker. Sorted offset `i` corresponds to leaf `leaf_base + i`. + by_start: Vec, + /// Active region/file identities mapped to leaf indices in `max_ends` for + /// removal without searching `by_start`. Entries are deleted when visited. + positions: HashMap, + /// Array-backed segment tree rooted at index 1 (index 0 is unused). + /// Leaves store inclusive file end times; internal nodes store the maximum + /// active end in their subtree. Removed/padding leaves and empty subtrees + /// hold `None`, allowing overlap queries to prune them. + max_ends: Vec>, + /// First leaf index in `max_ends`, also the padded leaf count: a power of two + /// with at least one leaf, even for an empty snapshot. + leaf_base: usize, +} + +struct OverlapQuery<'a> { + input: &'a FileHandle, + min_end: Timestamp, + candidates_end: usize, +} + +impl<'a> FileOverlapIndex<'a> { + pub(super) fn new(files: Vec, metadata: &'a RegionMetadata) -> Self { + let mut by_start: Vec<_> = (0..files.len()).collect(); + by_start.sort_unstable_by_key(|&i| (files[i].time_range().0, i)); + // A single empty leaf also handles an empty snapshot. + let leaf_base = files.len().next_power_of_two(); + let mut max_ends = vec![None; 2 * leaf_base]; + let mut positions = HashMap::with_capacity(files.len()); + for (offset, &i) in by_start.iter().enumerate() { + max_ends[leaf_base + offset] = Some(files[i].time_range().1); + positions.insert(files[i].file_id(), leaf_base + offset); + } + for node in (1..leaf_base).rev() { + max_ends[node] = max_ends[2 * node].max(max_ends[2 * node + 1]); + } + Self { + metadata, + files, + by_start, + positions, + max_ends, + leaf_base, + } + } + + pub(super) fn remove(&mut self, file_id: RegionFileId) { + let Some(mut node) = self.positions.remove(&file_id) else { + return; + }; + self.max_ends[node] = None; + while node > 1 { + node /= 2; + self.max_ends[node] = self.max_ends[2 * node].max(self.max_ends[2 * node + 1]); + } + } + + /// Removes dependencies so expansion never enumerates visited files. + /// Preserve snapshot order rather than exposing the index's time ordering + /// to the merge reader (including its handling of equal-sequence rows). + pub(super) fn drain_overlaps(&mut self, input: &FileHandle) -> Vec { + let mut matches = Vec::new(); + while let Some(i) = self.find_overlap(input) { + self.remove(self.files[i].file_id()); + matches.push(i); + } + matches.sort_unstable(); + matches.into_iter().map(|i| self.files[i].clone()).collect() + } + + fn find_overlap(&self, input: &FileHandle) -> Option { + let (start, end) = input.time_range(); + let query = OverlapQuery { + input, + min_end: start, + candidates_end: self + .by_start + .partition_point(|&i| self.files[i].time_range().0 <= end), + }; + self.find_in_subtree(1, 0..self.leaf_base, &query) + } + + fn find_in_subtree( + &self, + node: usize, + leaves: Range, + query: &OverlapQuery<'_>, + ) -> Option { + if leaves.start >= query.candidates_end + || self.max_ends[node].is_none_or(|end| end < query.min_end) + { + return None; + } + if leaves.len() == 1 { + let i = self.by_start[leaves.start]; + return files_may_overlap(query.input, &self.files[i], self.metadata).then_some(i); + } + let mid = leaves.start + leaves.len() / 2; + self.find_in_subtree(2 * node, leaves.start..mid, query) + .or_else(|| self.find_in_subtree(2 * node + 1, mid..leaves.end, query)) + } +} + +/// SST bounds are inclusive. Missing statistics, foreign encodings and schema +/// evolution must not exclude a possible logical-key dependency. +fn files_may_overlap(lhs: &FileHandle, rhs: &FileHandle, metadata: &RegionMetadata) -> bool { + let (lhs_start, lhs_end) = lhs.time_range(); + let (rhs_start, rhs_end) = rhs.time_range(); + if lhs_start.max(rhs_start) > lhs_end.min(rhs_end) { + return false; + } + + // FileMeta does not record the PK schema/encoding. Adding a tag can change + // encoded keys without changing the logical keys of old rows. + // TODO: Use schema-aware PK bounds before relaxing this fallback. Appending + // defaults to dense keys can make raw ranges miss logical overlaps, breaking + // closure picking and risking premature tombstone removal in TWCS. + if metadata.schema_version != 0 + || lhs.region_id() != metadata.region_id + || rhs.region_id() != metadata.region_id + { + return true; + } + match (lhs.primary_key_range(), rhs.primary_key_range()) { + (Some(lhs), Some(rhs)) if lhs.0 <= lhs.1 && rhs.0 <= rhs.1 => { + primary_key_ranges_overlap(&lhs, &rhs) + } + _ => true, + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use bytes::Bytes; + use rand::{Rng, SeedableRng}; + use store_api::storage::FileId; + + use super::*; + use crate::sst::file::FileMeta; + use crate::test_util::memtable_util::metadata_for_test; + use crate::test_util::new_noop_file_purger; + + fn file(start: i64, end: i64, pk: Option<(&str, &str)>) -> FileHandle { + FileHandle::new_with_primary_key_range( + FileMeta { + file_id: FileId::random(), + time_range: ( + Timestamp::new_millisecond(start), + Timestamp::new_millisecond(end), + ), + ..Default::default() + }, + new_noop_file_purger(), + pk.map(|(start, end)| { + ( + Bytes::copy_from_slice(start.as_bytes()), + Bytes::copy_from_slice(end.as_bytes()), + ) + }), + ) + } + + #[rstest::rstest] + #[case(10, Some(("b", "c")), 0, false, true)] + #[case(11, Some(("b", "c")), 0, false, false)] + #[case(10, Some(("c", "d")), 0, false, false)] + #[case(10, None, 0, false, true)] + #[case(10, Some(("z", "a")), 0, false, true)] + #[case(10, Some(("c", "d")), 1, false, true)] + #[case(10, Some(("c", "d")), 0, true, true)] + #[case(11, None, 1, true, false)] + fn test_conservative_overlap( + #[case] start: i64, + #[case] pk: Option<(&str, &str)>, + #[case] schema_version: u64, + #[case] foreign: bool, + #[case] expected: bool, + ) { + let mut metadata = (*metadata_for_test()).clone(); + metadata.region_id = 0.into(); + metadata.schema_version = schema_version; + let lhs = file(0, 10, Some(("a", "b"))); + let mut rhs = file(start, start + 10, pk); + if foreign { + let mut meta = rhs.meta_ref().clone(); + meta.region_id = 1.into(); + rhs = FileHandle::new_with_primary_key_range( + meta, + new_noop_file_purger(), + rhs.primary_key_range(), + ); + } + assert_eq!(expected, files_may_overlap(&lhs, &rhs, &metadata)); + assert_eq!(expected, files_may_overlap(&rhs, &lhs, &metadata)); + } + + #[test] + fn test_index_removal_distinguishes_region_owners() { + let metadata = metadata_for_test(); + let lhs = file(0, 10, None); + let mut meta = lhs.meta_ref().clone(); + meta.region_id = 1.into(); + let rhs = FileHandle::new(meta, new_noop_file_purger()); + let mut index = FileOverlapIndex::new(vec![lhs.clone(), rhs.clone()], &metadata); + index.remove(lhs.file_id()); + let overlaps = index.drain_overlaps(&lhs); + assert_eq!(1, overlaps.len()); + assert_eq!(rhs.file_id(), overlaps[0].file_id()); + assert!(index.drain_overlaps(&lhs).is_empty()); + } + + #[test] + fn test_index_matches_linear_scan_after_removals() { + let mut metadata = (*metadata_for_test()).clone(); + metadata.region_id = 0.into(); + let mut rng = rand::rngs::StdRng::seed_from_u64(9146); + for count in [0, 1, 7, 32, 127] { + let files: Vec<_> = (0..count) + .map(|_| { + let start = rng.random_range(-50..50); + let end = start + rng.random_range(0..40); + let pk = match rng.random_range(0..4) { + 0 => None, + 1 => Some(("a", "b")), + 2 => Some(("b", "c")), + _ => Some(("z", "z")), + }; + file(start, end, pk) + }) + .collect(); + let mut index = FileOverlapIndex::new(files.clone(), &metadata); + let mut removed = HashSet::new(); + for f in files.iter().step_by(3) { + index.remove(f.file_id()); + removed.insert(f.file_id()); + } + for _ in 0..32 { + let start = rng.random_range(-60..60); + let query = file(start, start + rng.random_range(0..25), Some(("b", "c"))); + let expected: Vec<_> = files + .iter() + .filter(|f| { + !removed.contains(&f.file_id()) && files_may_overlap(&query, f, &metadata) + }) + .map(FileHandle::file_id) + .collect(); + let actual: Vec<_> = index + .drain_overlaps(&query) + .iter() + .map(FileHandle::file_id) + .collect(); + assert_eq!(expected, actual, "snapshot size {count}"); + removed.extend(expected); + } + } + } +} diff --git a/src/mito2/src/compaction/picker.rs b/src/mito2/src/compaction/picker.rs index 693422955fa..de2e9d9d310 100644 --- a/src/mito2/src/compaction/picker.rs +++ b/src/mito2/src/compaction/picker.rs @@ -21,11 +21,12 @@ use common_time::{TimeToLive, Timestamp}; use serde::{Deserialize, Serialize}; use crate::compaction::compactor::CompactionRegion; +use crate::compaction::last_non_null::LastNonNullPicker; use crate::compaction::twcs::TwcsPicker; use crate::compaction::window::WindowedCompactionPicker; use crate::compaction::{CompactionOutput, SerializedCompactionOutput}; use crate::error::Result; -use crate::region::options::CompactionOptions; +use crate::region::options::{CompactionOptions, MergeMode, RegionOptions}; use crate::sst::file::{FileHandle, FileMeta}; use crate::sst::file_purger::FilePurger; use crate::sst::version::LevelMeta; @@ -125,11 +126,10 @@ impl PickerOutput { } } -/// Create a new picker based on the compaction request options and compaction options. +/// Creates a picker for the request and the region's compaction and merge modes. pub fn new_picker( compact_request_options: &compact_request::Options, - compaction_options: &CompactionOptions, - append_mode: bool, + region_options: &RegionOptions, max_background_tasks: Option, time_range: Option, ) -> Arc { @@ -139,20 +139,29 @@ pub fn new_picker( } else { Some(window.window_seconds) }; + // Strict-window outputs already include every overlapping file segment + // within each disjoint, half-open output range, including LastNonNull. Arc::new(WindowedCompactionPicker::new(window).with_time_range(time_range)) as Arc<_> } else { - match compaction_options { - CompactionOptions::Twcs(twcs_opts) => Arc::new(TwcsPicker { + let picker = match ®ion_options.compaction { + CompactionOptions::Twcs(twcs_opts) => TwcsPicker { trigger_file_num: twcs_opts.active_window_trigger_file_num, active_window_l1_merge_trigger: twcs_opts.active_window_l1_merge_trigger, inactive_window_trigger_file_num: twcs_opts.inactive_window_trigger_file_num, inactive_window_l1_merge_trigger: twcs_opts.inactive_window_l1_merge_trigger, time_window_seconds: twcs_opts.time_window_seconds(), max_output_file_size: twcs_opts.max_output_file_size.map(|r| r.as_bytes()), - append_mode, + append_mode: region_options.append_mode, max_background_tasks, time_range, - }) as Arc<_>, + }, + }; + if region_options.merge_mode() == MergeMode::LastNonNull { + // LastNonNull correctness spans windows and levels, so wrap the seed + // picker instead of trusting TWCS's independent windows. + Arc::new(LastNonNullPicker::new(picker)) + } else { + Arc::new(picker) } } } diff --git a/src/mito2/src/compaction/scheduler/planning.rs b/src/mito2/src/compaction/scheduler/planning.rs index facb1bb1d3d..23318ad7669 100644 --- a/src/mito2/src/compaction/scheduler/planning.rs +++ b/src/mito2/src/compaction/scheduler/planning.rs @@ -206,10 +206,13 @@ impl CompactionScheduler { // Avoid queuing serial outputs from a stale snapshot. Repicking after each // batch lets newly flushed L0 files outrank L1 work that has not started. let max_picker_outputs = max_background_compactions.min(request.max_parallelism.max(1)); + let region_options = RegionOptions { + compaction: dynamic_compaction_opts, + ..request.current_version.options.clone() + }; let picker = new_picker( &options, - &dynamic_compaction_opts, - request.current_version.options.append_mode, + ®ion_options, Some(max_picker_outputs), time_range, ); @@ -235,10 +238,7 @@ impl CompactionScheduler { let compaction_region = CompactionRegion { region_id, current_version: current_version.clone(), - region_options: RegionOptions { - compaction: dynamic_compaction_opts.clone(), - ..current_version.options.clone() - }, + region_options, engine_config: engine_config.clone(), region_metadata: current_version.metadata.clone(), cache_manager: cache_manager.clone(), diff --git a/src/mito2/src/compaction/test_util.rs b/src/mito2/src/compaction/test_util.rs index f6b39fdc8cf..453af8a8a68 100644 --- a/src/mito2/src/compaction/test_util.rs +++ b/src/mito2/src/compaction/test_util.rs @@ -13,13 +13,23 @@ // limitations under the License. use std::num::NonZeroU64; +use std::sync::Arc; +use std::time::Duration; use bytes::Bytes; +use common_base::Plugins; use common_time::Timestamp; use store_api::storage::FileId; +use crate::cache::CacheManager; +use crate::compaction::compactor::{CompactionRegion, CompactionVersion}; +use crate::config::MitoConfig; +use crate::region::options::RegionOptions; use crate::sst::file::{FileHandle, FileMeta, Level}; +use crate::sst::version::SstVersion; +use crate::test_util::memtable_util::metadata_for_test; use crate::test_util::new_noop_file_purger; +use crate::test_util::scheduler_util::SchedulerEnv; /// Test util to create file handles. pub fn new_file_handle( @@ -128,3 +138,38 @@ pub fn new_file_handle_with_size_sequence_and_primary_key_range( primary_key_range, ) } + +pub(crate) async fn compaction_region_with_ssts( + files: impl IntoIterator, + ttl: Duration, +) -> CompactionRegion { + let env = SchedulerEnv::new().await; + let metadata = metadata_for_test(); + let manifest_ctx = env.mock_manifest_context(metadata.clone()).await; + let mut ssts = SstVersion::new(); + ssts.add_files( + Arc::new(crate::sst::file_purger::NoopFilePurger), + files.into_iter(), + ); + + CompactionRegion { + region_id: metadata.region_id, + region_options: RegionOptions::default(), + engine_config: Arc::new(MitoConfig::default()), + region_metadata: metadata.clone(), + cache_manager: Arc::new(CacheManager::default()), + access_layer: env.access_layer, + manifest_ctx, + current_version: CompactionVersion { + metadata, + options: RegionOptions::default(), + ssts: Arc::new(ssts), + memtable_min_sequence: None, + compaction_time_window: None, + }, + file_purger: None, + ttl: Some(ttl.into()), + max_parallelism: 1, + plugins: Plugins::new(), + } +} diff --git a/src/mito2/src/compaction/twcs.rs b/src/mito2/src/compaction/twcs.rs index 2953ded8968..b2e3274b062 100644 --- a/src/mito2/src/compaction/twcs.rs +++ b/src/mito2/src/compaction/twcs.rs @@ -71,6 +71,12 @@ struct WindowPickContext<'a> { phase: PickPhase, } +struct WindowOutputContext { + active_window: Option, + time_window_size: Option, + max_outputs: Option, +} + /// A mixed L0/L1 compaction may rewrite at most this many L1 rows per L0 row. const MAX_L1_L0_ROW_RATIO: usize = 2; @@ -119,13 +125,17 @@ pub struct TwcsPicker { } impl TwcsPicker { - async fn build_output_with_time_range( + async fn build_output( &self, region_id: RegionId, time_windows: BTreeMap, - active_window: Option, - time_window_size: Option, + context: WindowOutputContext, ) -> Result> { + let WindowOutputContext { + active_window, + time_window_size, + max_outputs, + } = context; let mut output = vec![]; let windows = time_windows .values() @@ -145,6 +155,8 @@ impl TwcsPicker { .map(|window| window.time_window) .collect::>(); let time_windows = Arc::new(time_windows); + // Enumerating all LastNonNull seeds must not increase the number of + // concurrently planned windows along with the output limit. let chunk_size = self.max_background_tasks.unwrap_or(windows.len()).max(1); let mut selected_windows = HashSet::new(); 'phases: for phase in PICK_PHASES { @@ -198,12 +210,12 @@ impl TwcsPicker { output_time_range: None, // we do not enforce output time range in twcs compactions. }); - if let Some(max_background_tasks) = self.max_background_tasks - && output.len() >= max_background_tasks + if let Some(max_outputs) = max_outputs + && output.len() >= max_outputs { debug!( - "Region ({:?}) compaction task size larger than max background tasks({}), remaining tasks discarded", - region_id, max_background_tasks + "Region ({:?}) compaction output limit ({}) reached, remaining candidates discarded", + region_id, max_outputs ); break 'phases; } @@ -772,6 +784,17 @@ fn log_pick_result( #[async_trait::async_trait] impl Picker for TwcsPicker { async fn pick(&self, compaction_region: &CompactionRegion) -> Result> { + self.pick_with_output_limit(compaction_region, self.max_background_tasks) + .await + } +} + +impl TwcsPicker { + pub(crate) async fn pick_with_output_limit( + &self, + compaction_region: &CompactionRegion, + max_outputs: Option, + ) -> Result> { let region_id = compaction_region.region_id; let picker = self.clone(); let compaction_region = compaction_region.clone(); @@ -835,7 +858,15 @@ impl Picker for TwcsPicker { .context(JoinSnafu)?; let outputs = self - .build_output_with_time_range(region_id, windows, active_window, Some(time_window_size)) + .build_output( + region_id, + windows, + WindowOutputContext { + active_window, + time_window_size: Some(time_window_size), + max_outputs, + }, + ) .await?; if outputs.is_empty() && expired_ssts.is_empty() { @@ -1042,7 +1073,8 @@ mod tests { use crate::cache::CacheManager; use crate::compaction::compactor::CompactionVersion; use crate::compaction::test_util::{ - new_file_handle, new_file_handle_with_sequence, new_file_handle_with_size_and_sequence, + compaction_region_with_ssts, new_file_handle, new_file_handle_with_sequence, + new_file_handle_with_size_and_sequence, new_file_handle_with_size_sequence_and_primary_key_range, }; use crate::config::MitoConfig; @@ -1052,6 +1084,27 @@ mod tests { use crate::test_util::memtable_util::metadata_for_test; use crate::test_util::scheduler_util::SchedulerEnv; + impl TwcsPicker { + async fn build_output_with_time_range( + &self, + region_id: RegionId, + time_windows: BTreeMap, + active_window: Option, + time_window_size: Option, + ) -> Result> { + self.build_output( + region_id, + time_windows, + WindowOutputContext { + active_window, + time_window_size, + max_outputs: self.max_background_tasks, + }, + ) + .await + } + } + #[test] fn test_valid_max_input_files_env_overrides_default() { assert_eq!(64, parse_max_input_files(Some("64"))); @@ -1064,40 +1117,6 @@ mod tests { } } - async fn compaction_region_with_ssts( - files: impl IntoIterator, - ttl: Duration, - ) -> CompactionRegion { - let env = SchedulerEnv::new().await; - let metadata = metadata_for_test(); - let manifest_ctx = env.mock_manifest_context(metadata.clone()).await; - let mut ssts = SstVersion::new(); - ssts.add_files( - Arc::new(crate::sst::file_purger::NoopFilePurger), - files.into_iter(), - ); - - CompactionRegion { - region_id: metadata.region_id, - region_options: RegionOptions::default(), - engine_config: Arc::new(MitoConfig::default()), - region_metadata: metadata.clone(), - cache_manager: Arc::new(CacheManager::default()), - access_layer: env.access_layer, - manifest_ctx, - current_version: CompactionVersion { - metadata, - options: RegionOptions::default(), - ssts: Arc::new(ssts), - compaction_time_window: None, - }, - file_purger: None, - ttl: Some(ttl.into()), - max_parallelism: 1, - plugins: Plugins::new(), - } - } - async fn compaction_region_with_expired_sst() -> CompactionRegion { compaction_region_with_ssts( (1..=4).map(|sequence| FileMeta { @@ -1259,6 +1278,7 @@ mod tests { metadata, options: RegionOptions::default(), ssts: Arc::new(ssts), + memtable_min_sequence: None, compaction_time_window: None, }, file_purger: None, diff --git a/src/mito2/src/compaction/window.rs b/src/mito2/src/compaction/window.rs index a9f3f021b84..702dffeb8f8 100644 --- a/src/mito2/src/compaction/window.rs +++ b/src/mito2/src/compaction/window.rs @@ -26,8 +26,10 @@ use store_api::storage::RegionId; use crate::compaction::CompactionOutput; use crate::compaction::buckets::infer_time_bucket; use crate::compaction::compactor::{CompactionRegion, CompactionVersion}; +use crate::compaction::last_non_null::inputs_precede_memtables; use crate::compaction::picker::{Picker, PickerOutput, get_expired_ssts}; use crate::error::{JoinSnafu, Result}; +use crate::region::options::MergeMode; use crate::sst::file::FileHandle; /// Compaction picker that splits the time range of all involved files to windows, and merges @@ -102,6 +104,9 @@ impl WindowedCompactionPicker { .map(|file| file.file_id()) .collect::>(); + let last_non_null = current_version.options.merge_mode() == MergeMode::LastNonNull; + // Busy SSTs remain LastNonNull dependencies: skipping an intermediate + // version would allow older fields to be promoted above it. let windows = assign_files_to_time_windows( time_window, current_version @@ -109,10 +114,22 @@ impl WindowedCompactionPicker { .levels() .iter() .flat_map(|level| level.files.values()) - .filter(|file| !expired_file_ids.contains(&file.file_id())), + .filter(|file| !expired_file_ids.contains(&file.file_id())) + .filter(|file| last_non_null || !file.compacting()), ); let windows = filter_time_windows(windows, self.time_range); + // One SST can span several outputs. Deferring only an unsafe window + // would remove that SST without rewriting all its remaining segments. + if last_non_null + && windows.values().any(|(_, inputs)| { + inputs.iter().any(FileHandle::compacting) + || !inputs_precede_memtables(inputs, current_version.memtable_min_sequence) + }) + { + return (vec![], expired_ssts, time_window); + } + (build_output(windows), expired_ssts, time_window) } } @@ -120,6 +137,11 @@ impl WindowedCompactionPicker { #[async_trait::async_trait] impl Picker for WindowedCompactionPicker { async fn pick(&self, compaction_region: &CompactionRegion) -> Result> { + // TODO: With per-region compaction concurrency, the scheduler must wait + // for all running compactions to finish and apply their SST edits, and + // drain or cancel queued compactions before capturing current_version + // for SWCS. Block new compactions through snapshot capture and picking + // so the strict-window plan sees the complete SST view. let picker = self.clone(); let region_id = compaction_region.current_version.metadata.region_id; let current_version = compaction_region.current_version.clone(); @@ -233,9 +255,6 @@ fn assign_files_to_time_windows<'a>( let mut buckets = BTreeMap::new(); for file in files { - if file.compacting() { - continue; - } let (start, end) = file.time_range(); let bounds = file_time_bucket_span( // safety: converting whatever timestamp to seconds will not overflow. @@ -280,6 +299,7 @@ fn file_time_bucket_span(start_sec: i64, end_sec: i64, bucket_sec: i64) -> Vec<( #[cfg(test)] mod tests { + use std::collections::HashSet; use std::sync::Arc; use std::time::Duration; @@ -289,7 +309,7 @@ mod tests { use crate::compaction::compactor::CompactionVersion; use crate::compaction::window::{WindowedCompactionPicker, file_time_bucket_span}; - use crate::region::options::RegionOptions; + use crate::region::options::{MergeMode, RegionOptions}; use crate::sst::file::{FileMeta, Level}; use crate::sst::file_purger::NoopFilePurger; use crate::sst::version::SstVersion; @@ -320,6 +340,7 @@ mod tests { CompactionVersion { metadata, ssts: Arc::new(ssts), + memtable_min_sequence: None, options: RegionOptions { ttl: ttl.map(|t| t.into()), auto_flush_interval: None, @@ -516,6 +537,75 @@ mod tests { assert!(outputs.is_empty()); } + #[rstest::rstest] + #[case::requested_window(0)] + #[case::transitive_window(1)] + #[case::unrelated_window(3)] + fn test_busy_dependencies_defer_last_non_null_plan_until_released( + #[case] busy_window: i64, + #[values(MergeMode::LastRow, MergeMode::LastNonNull)] merge_mode: MergeMode, + ) { + // A and C span both output windows. Skipping a busy intermediate B + // would allow an unsafe A+C merge; deferring just one output would + // lose the other segment of these shared inputs when they are removed. + let [a, b, c, expired] = std::array::from_fn(|_| FileId::random()); + let files = [ + (a, 0, 2 * HOUR - 1, 0), + (b, busy_window * HOUR, (busy_window + 1) * HOUR - 1, 1), + (c, 0, 2 * HOUR - 1, 0), + (expired, -2 * HOUR, -2 * HOUR, 0), + ]; + let mut version = build_version(&files, Some(Duration::from_millis(3 * HOUR as u64))); + version.options.merge_mode = Some(merge_mode); + // No memtable barrier: only the busy SST dependency may defer this plan. + assert_eq!(None, version.memtable_min_sequence); + let busy_file = version.ssts.levels()[1].files().next().unwrap(); + let picker = + WindowedCompactionPicker::new(Some(HOUR / 1000)).with_time_range(TimestampRange::new( + Timestamp::new_millisecond(0), + Timestamp::new_millisecond(HOUR), + )); + + for busy in [true, false] { + busy_file.set_compacting(busy); + let (outputs, expired_ssts, _) = picker.pick_inner( + version.metadata.region_id, + &version, + Timestamp::new_millisecond(2 * HOUR), + ); + assert_eq!(1, expired_ssts.len()); + assert_eq!(expired, expired_ssts[0].meta_ref().file_id); + assert!(!expired_ssts[0].compacting()); + + if busy && merge_mode == MergeMode::LastNonNull && busy_window < 2 { + assert!(outputs.is_empty(), "the entire dependent plan must wait"); + continue; + } + + assert_eq!(2, outputs.len()); + for (window, output) in outputs.iter().enumerate() { + let window = window as i64; + let mut expected = HashSet::from([a, c]); + if !busy && busy_window == window { + expected.insert(b); + } + let actual: HashSet<_> = output + .inputs + .iter() + .map(|file| file.meta_ref().file_id) + .collect(); + assert_eq!(expected, actual); + assert_eq!( + TimestampRange::new( + Timestamp::new_millisecond(window * HOUR), + Timestamp::new_millisecond((window + 1) * HOUR), + ), + output.output_time_range + ); + } + } + } + #[test] fn test_file_time_bucket_span() { assert_eq!( diff --git a/src/mito2/src/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index 8d45f852592..07073ca6a6a 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -28,6 +28,7 @@ use common_error::ext::{ErrorExt, RetryHint}; use common_recordbatch::RecordBatches; use common_time::util::current_time_millis; use common_wal::options::{KafkaWalOptions, WAL_OPTIONS_KEY, WalOptions}; +use datafusion_expr::{col, lit}; use rstest::rstest; use rstest_reuse::{self, apply}; use store_api::ManifestVersion; @@ -35,9 +36,9 @@ use store_api::metadata::RegionMetadataRef; use store_api::mito_engine_options::WRITE_BUFFER_SIZE_KEY; use store_api::region_engine::{RegionEngine, RegionRole}; use store_api::region_request::{ - AlterKind, PathType, RegionAlterRequest, RegionCloseRequest, RegionFlushRequest, - RegionOpenRequest, RegionPutRequest, RegionRequest, ReplayCheckpoint, SetRegionOption, - UnsetRegionOption, + AlterKind, PathType, RegionAlterRequest, RegionCloseRequest, RegionCompactRequest, + RegionFlushRequest, RegionOpenRequest, RegionPutRequest, RegionRequest, ReplayCheckpoint, + SetRegionOption, UnsetRegionOption, }; use store_api::storage::{RegionId, ScanRequest}; use tokio::sync::Notify; @@ -49,9 +50,11 @@ use crate::engine::listener::{EventListener, FlushListener, StallListener}; use crate::engine::region_hook::{RegionGcInfo, RegionHook, RegionHookRef, SstFileInfo}; use crate::error::Error; use crate::manifest::action::RegionMetaActionList; +use crate::memtable::RangesOptions; use crate::test_util::{ - CreateRequestBuilder, LogStoreFactory, MockWriteBufferManager, TestEnv, build_rows, - build_rows_for_key, flush_region, kafka_log_store_factory, multiple_log_store_factories, + CreateRequestBuilder, LogStoreFactory, MockWriteBufferManager, TestEnv, + build_delete_rows_for_key, build_rows, build_rows_for_key, build_rows_with_fields, delete_rows, + delete_rows_schema, flush_region, kafka_log_store_factory, multiple_log_store_factories, prepare_test_for_kafka_log_store, put_rows, raft_engine_log_store_factory, reopen_region, rows_schema, single_kafka_log_store_factory, }; @@ -1820,3 +1823,299 @@ async fn test_region_hook() { manifest_count, ); } + +/// Explicitly finish a merge instead of sleeping for the background dispatcher. +fn compact_bulk_memtables(engine: &MitoEngine, region_id: RegionId) { + let version = engine.get_region(region_id).unwrap().version(); + for memtable in version.memtables.list_memtables() { + memtable.compact(false).unwrap(); + } +} + +async fn scan_bulk_test_keys(engine: &MitoEngine, region_id: RegionId) -> String { + let mut schema = None; + let mut batches = Vec::new(); + for key in ["deleted", "k"] { + let request = ScanRequest { + filters: vec![col("tag_0").eq(lit(key))], + ..Default::default() + }; + let stream = engine.scan_to_stream(region_id, request).await.unwrap(); + schema = Some(stream.schema()); + batches.extend(RecordBatches::try_collect(stream).await.unwrap().take()); + } + RecordBatches::try_new(schema.unwrap(), batches) + .unwrap() + .pretty_print() + .unwrap() +} + +#[rstest] +#[tokio::test] +async fn test_bulk_flush_keeps_newer_version(#[values(false, true)] delete_newer: bool) { + let mut env = TestEnv::new().await; + let config = MitoConfig { + default_flat_format: true, + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new() + .insert_option("memtable.type", "bulk") + .insert_option("memtable.bulk.merge_threshold", "2") + .insert_option("memtable.bulk.encode_row_threshold", "1") + .build(); + let schema = rows_schema(&request); + let delete_schema = delete_rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await + .unwrap(); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: build_rows_for_key("k", 1, 2, 1), + }, + ) + .await; + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: build_rows_for_key("a", 0, 1024, 0), + }, + ) + .await; + if delete_newer { + let mut rows = build_delete_rows_for_key("b", 0, 1023); + rows.extend(build_delete_rows_for_key("k", 1, 2)); + delete_rows( + &engine, + region_id, + Rows { + schema: delete_schema, + rows, + }, + ) + .await; + } else { + let mut rows = build_rows_for_key("b", 0, 1023, 0); + rows.extend(build_rows_for_key("k", 1, 2, 2)); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows, + }, + ) + .await; + } + compact_bulk_memtables(&engine, region_id); + // Previously this advanced the old unordered row above the newer encoded row. + put_rows( + &engine, + region_id, + Rows { + schema, + rows: build_rows_for_key("c", 0, 1, 0), + }, + ) + .await; + let expected = if delete_newer { + "++\n++" + } else { + "\ ++-------+---------+---------------------+ +| tag_0 | field_0 | ts | ++-------+---------+---------------------+ +| k | 2.0 | 1970-01-01T00:00:01 | ++-------+---------+---------------------+" + }; + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); + flush_region(&engine, region_id, None).await; + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); + let engine = env.reopen_engine(engine, config).await; + reopen_region( + &engine, + region_id, + request.table_dir, + false, + request.options, + ) + .await; + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); +} + +#[rstest] +#[tokio::test] +async fn test_bulk_last_non_null_merge_and_flush(#[values(false, true)] encoded: bool) { + let mut env = TestEnv::new().await; + let config = MitoConfig { + default_flat_format: true, + min_compaction_interval: Duration::from_secs(3600), + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; + let region_id = RegionId::new(1, 1); + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + let request = CreateRequestBuilder::new() + .field_num(2) + .insert_option("memtable.type", "bulk") + .insert_option("merge_mode", "last_non_null") + .insert_option("memtable.bulk.merge_threshold", "4") + .insert_option( + "memtable.bulk.encode_row_threshold", + if encoded { "1" } else { "1000000" }, + ) + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.trigger_file_num", "2") + .build(); + let schema = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request.clone())) + .await + .unwrap(); + // An older SST must remain hidden behind a delete retained in the memtable. + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: build_rows_with_fields("deleted", &[1], &[(Some(1), Some(1))]), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + + let mut a = build_rows_with_fields( + "a", + &(0..1023).collect::>(), + &[(Some(0), None); 1023], + ); + a.extend(build_rows_with_fields("k", &[1], &[(Some(1), None)])); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: a, + }, + ) + .await; + // Intermediate writes must be bulk parts, not an unordered merge barrier. + let mut b = build_rows_with_fields( + "b", + &(0..1023).collect::>(), + &[(Some(0), None); 1023], + ); + b.extend(build_rows_with_fields("k", &[1], &[(Some(2), None)])); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: b, + }, + ) + .await; + let mut deletes = build_delete_rows_for_key("d", 0, 1023); + deletes.extend(build_delete_rows_for_key("deleted", 1, 2)); + delete_rows( + &engine, + region_id, + Rows { + schema: delete_rows_schema(&request), + rows: deletes, + }, + ) + .await; + let before = scan_bulk_test_keys(&engine, region_id).await; + assert!(before.contains("| k | 2.0")); + assert!(!before.contains("| deleted")); + + let mut c = build_rows_with_fields( + "c", + &(0..1022).collect::>(), + &[(Some(0), None); 1022], + ); + c.extend(build_rows_with_fields("k", &[1], &[(None, Some(3))])); + c.extend(build_rows_with_fields("deleted", &[1], &[(None, Some(3))])); + put_rows( + &engine, + region_id, + Rows { + schema: schema.clone(), + rows: c, + }, + ) + .await; + compact_bulk_memtables(&engine, region_id); + // All four sources must have merged, with the requested output format. + let version = engine.get_region(region_id).unwrap().version(); + let mut ranges = Vec::new(); + for memtable in version.memtables.list_memtables() { + ranges.extend( + memtable + .ranges(None, RangesOptions::default()) + .unwrap() + .ranges + .into_values(), + ); + } + assert_eq!(1, ranges.len()); + assert_eq!(encoded, ranges[0].encoded().is_some()); + drop(ranges); + drop(version); + let expected = "\ ++---------+---------+---------+---------------------+ +| tag_0 | field_0 | field_1 | ts | ++---------+---------+---------+---------------------+ +| deleted | | 3.0 | 1970-01-01T00:00:01 | +| k | 2.0 | 3.0 | 1970-01-01T00:00:01 | ++---------+---------+---------+---------------------+"; + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); + // Leave a raw source beside merged parts so flush also exercises grouping. + put_rows( + &engine, + region_id, + Rows { + schema, + rows: build_rows_with_fields("z", &[0], &[(Some(0), None)]), + }, + ) + .await; + flush_region(&engine, region_id, None).await; + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); + engine + .handle_request( + region_id, + RegionRequest::Compact(RegionCompactRequest::default()), + ) + .await + .unwrap(); + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); + let engine = env.reopen_engine(engine, config).await; + reopen_region( + &engine, + region_id, + request.table_dir, + false, + request.options, + ) + .await; + assert_eq!(expected, scan_bulk_test_keys(&engine, region_id).await); +} diff --git a/src/mito2/src/engine/merge_mode_test.rs b/src/mito2/src/engine/merge_mode_test.rs index 5430e6b246c..f5dd078103d 100644 --- a/src/mito2/src/engine/merge_mode_test.rs +++ b/src/mito2/src/engine/merge_mode_test.rs @@ -23,7 +23,9 @@ use store_api::region_engine::RegionEngine; use store_api::region_request::{RegionCompactRequest, RegionRequest}; use store_api::storage::{RegionId, ScanRequest}; +use crate::compaction::compactor::CompactionVersion; use crate::config::MitoConfig; +use crate::manifest::action::RegionEdit; use crate::test_util::batch_util::sort_batches_and_print; use crate::test_util::{ CreateRequestBuilder, TestEnv, build_delete_rows_for_key, build_rows_with_fields, delete_rows, @@ -32,8 +34,220 @@ use crate::test_util::{ #[rstest::rstest] #[tokio::test] -async fn test_partial_compaction_preserves_last_row_put_versions( +async fn test_admitted_sst_compaction_preserves_intermediate_memtable( #[values(false, true)] flat_format: bool, + #[values(false, true)] strict_window: bool, + #[values(false, true)] immutable: bool, +) { + // Issue #9157: A@1 and admitted C@3 must not merge across memtable B@2. + let mut env = TestEnv::new().await; + let config = MitoConfig { + default_flat_format: flat_format, + schedule_compaction_after_edit: false, + min_compaction_interval: std::time::Duration::from_secs(3600), + ..Default::default() + }; + let engine = env.create_engine(config.clone()).await; + let region_id = RegionId::new(1, 1); + let source_id = RegionId::new(1, 2); + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + let request = CreateRequestBuilder::new() + .field_num(2) + .insert_option("compaction.type", "twcs") + .insert_option("compaction.twcs.time_window", "1h") + .insert_option("compaction.twcs.trigger_file_num", "2") + .insert_option("merge_mode", "last_non_null") + .build(); + let table_dir = request.table_dir.clone(); + let region_opts = request.options.clone(); + let schema = rows_schema(&request); + engine + .handle_request(source_id, RegionRequest::Create(request.clone())) + .await + .unwrap(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + for (id, fields) in [(region_id, (Some(1), None)), (source_id, (None, Some(3)))] { + put_rows( + &engine, + id, + Rows { + schema: schema.clone(), + rows: build_rows_with_fields("a", &[10], &[fields]), + }, + ) + .await; + flush_region(&engine, id, None).await; + } + put_rows( + &engine, + region_id, + Rows { + schema, + rows: build_rows_with_fields("a", &[10], &[(Some(2), None)]), + }, + ) + .await; + let region = engine.get_region(region_id).unwrap(); + if immutable { + region.version_control.freeze_mutable().unwrap(); + } + let foreign = engine + .get_region(source_id) + .unwrap() + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files()) + .next() + .unwrap() + .meta_ref() + .clone(); + assert_eq!(source_id, foreign.region_id); + engine + .edit_region( + region_id, + RegionEdit { + files_to_add: vec![foreign], + files_to_remove: vec![], + timestamp_ms: None, + compaction_time_window: None, + flushed_entry_id: None, + flushed_sequence: None, + committed_sequence: None, + }, + ) + .await + .unwrap(); + assert_eq!(3, region.version_control.committed_sequence()); + let before_ids = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap() + .file_ids(); + assert_eq!(2, before_ids.len()); + let expected = "\ ++-------+---------+---------+---------------------+ +| tag_0 | field_0 | field_1 | ts | ++-------+---------+---------+---------------------+ +| a | 2.0 | 3.0 | 1970-01-01T00:00:10 | ++-------+---------+---------+---------------------+"; + let before = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(expected, before); + let compact_request = || { + if strict_window { + RegionCompactRequest { + options: api::v1::region::compact_request::Options::StrictWindow( + api::v1::region::StrictWindow { + window_seconds: 3600, + }, + ), + ..Default::default() + } + } else { + RegionCompactRequest::default() + } + }; + engine + .handle_request(region_id, RegionRequest::Compact(compact_request())) + .await + .unwrap(); + let after = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!( + expected, after, + "compaction must not promote A's x above memtable B" + ); + let after_ids = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap() + .file_ids(); + assert_eq!(before_ids, after_ids, "the unsafe closure must be deferred"); + + let old_version = region.version(); + let snapshot = CompactionVersion::from(old_version.clone()); + assert_eq!(Some(2), snapshot.memtable_min_sequence); + flush_region(&engine, region_id, None).await; + // Flush publishes a new Version. Reusing the old snapshot cannot observe + // empty memtables paired with its pre-flush SST set. + assert_eq!( + Some(2), + CompactionVersion::from(old_version).memtable_min_sequence + ); + assert_eq!(Some(2), snapshot.memtable_min_sequence); + assert_eq!( + None, + CompactionVersion::from(region.version()).memtable_min_sequence + ); + engine + .handle_request(region_id, RegionRequest::Compact(compact_request())) + .await + .unwrap(); + let scanner = engine + .scanner(region_id, ScanRequest::default()) + .await + .unwrap(); + assert_eq!( + 1, + scanner.num_files(), + "flush makes the complete A+B+C merge eligible" + ); + let merged = RecordBatches::try_collect(scanner.scan().await.unwrap()) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(expected, merged); + let engine = env.reopen_engine(engine, config).await; + reopen_region(&engine, region_id, table_dir, false, region_opts).await; + let reopened = RecordBatches::try_collect( + engine + .scan_to_stream(region_id, ScanRequest::default()) + .await + .unwrap(), + ) + .await + .unwrap() + .pretty_print() + .unwrap(); + assert_eq!(expected, reopened); +} + +#[rstest::rstest] +#[tokio::test] +async fn test_partial_compaction_preserves_put_versions( + #[values(false, true)] flat_format: bool, + #[values("last_row", "last_non_null")] merge_mode: &str, ) { let mut env = TestEnv::new().await; let config = MitoConfig { @@ -57,7 +271,7 @@ async fn test_partial_compaction_preserves_last_row_put_versions( .field_num(2) .insert_option("compaction.type", "twcs") .insert_option("compaction.twcs.time_window", "1h") - .insert_option("merge_mode", "last_row") + .insert_option("merge_mode", merge_mode) .build(); let table_dir = request.table_dir.clone(); let region_opts = request.options.clone(); @@ -67,8 +281,9 @@ async fn test_partial_compaction_preserves_last_row_put_versions( .await .unwrap(); - // A(x=1), B(x=2) share a key and timestamp; C is unrelated. - // Promoting A when merging A+C must not hide the unselected B. + // Issue #9146: A(x=1), B(x=2) share a key and timestamp. For LastNonNull, + // C(y=3) also shares that key: merging A+C must not promote A's x above B. + // For LastRow, C is unrelated and B must remain outside the partial pick. let a = build_rows_with_fields("a", &[10], &[(Some(1), None)]); put_rows( &engine, @@ -110,7 +325,12 @@ async fn test_partial_compaction_preserves_last_row_put_versions( .find(|file| file.meta_ref().num_rows == 3001) .unwrap() .file_id(); - let c = build_rows_with_fields("z", &[10], &[(None, Some(3))]); + let last_non_null = merge_mode == "last_non_null"; + let c = build_rows_with_fields( + if last_non_null { "a" } else { "z" }, + &[10], + &[(None, Some(3))], + ); put_rows( &engine, region_id, @@ -146,12 +366,21 @@ async fn test_partial_compaction_preserves_last_row_put_versions( .unwrap() .pretty_print() .unwrap(); - let expected = "\ + let expected = if last_non_null { + "\ ++-------+---------+---------+---------------------+ +| tag_0 | field_0 | field_1 | ts | ++-------+---------+---------+---------------------+ +| a | 2.0 | 3.0 | 1970-01-01T00:00:10 | ++-------+---------+---------+---------------------+" + } else { + "\ +-------+---------+---------+---------------------+ | tag_0 | field_0 | field_1 | ts | +-------+---------+---------+---------------------+ | a | 2.0 | | 1970-01-01T00:00:10 | -+-------+---------+---------+---------------------+"; ++-------+---------+---------+---------------------+" + }; assert_eq!(expected, before); engine .handle_request( @@ -165,7 +394,10 @@ async fn test_partial_compaction_preserves_last_row_put_versions( .unwrap() .pretty_print() .unwrap(); - assert_eq!(expected, after, "flat_format={flat_format}"); + assert_eq!( + expected, after, + "flat_format={flat_format}, mode={merge_mode}" + ); let version = engine.get_region(region_id).unwrap().version(); let files: Vec<_> = version .ssts @@ -173,8 +405,11 @@ async fn test_partial_compaction_preserves_last_row_put_versions( .iter() .flat_map(|level| level.files()) .collect(); - assert_eq!(2, files.len()); - assert!(files.iter().any(|file| file.file_id() == b_id)); + assert_eq!(if last_non_null { 1 } else { 2 }, files.len()); + assert_eq!( + !last_non_null, + files.iter().any(|file| file.file_id() == b_id) + ); assert!( files .iter() diff --git a/src/mito2/src/flush.rs b/src/mito2/src/flush.rs index 883a44845aa..6599ccbe2bb 100644 --- a/src/mito2/src/flush.rs +++ b/src/mito2/src/flush.rs @@ -46,7 +46,10 @@ use crate::error::{ }; use crate::manifest::action::{RegionEdit, RegionMetaAction, RegionMetaActionList}; use crate::memtable::bulk::ENCODE_ROW_THRESHOLD; -use crate::memtable::{BoxedRecordBatchIterator, EncodedRange, MemtableRanges, RangesOptions}; +use crate::memtable::{ + BoxedRecordBatchIterator, EncodedRange, MemtableRange, MemtableRanges, RangesOptions, + group_by_sequence, +}; use crate::metrics::{ FLUSH_BYTES_TOTAL, FLUSH_ELAPSED, FLUSH_FAILURE_TOTAL, FLUSH_FILE_TOTAL, FLUSH_REQUESTS_TOTAL, INFLIGHT_FLUSH_COUNT, @@ -836,7 +839,10 @@ impl RegionFlushTask { source, cache_manager: self.cache_manager.clone(), storage: version.options.storage.clone(), - max_sequence: Some(max_sequence), + // Sources have disjoint sequence intervals. LastNonNull can still + // contain Puts and retained delete barriers for the same row. + max_sequence: (version.options.merge_mode() != MergeMode::LastNonNull) + .then_some(max_sequence), sst_write_format: if flat_format { FormatType::Flat } else { @@ -908,7 +914,7 @@ struct FlatSources { encoded: SmallVec<[(EncodedRange, SequenceNumber); 4]>, } -/// Returns the max sequence and [FlatSource] for the given memtable. +/// Builds sequence-disjoint sources and their maximum input sequences. fn memtable_flat_sources( schema: SchemaRef, mem_ranges: MemtableRanges, @@ -916,190 +922,158 @@ fn memtable_flat_sources( options: &RegionOptions, field_column_start: usize, ) -> Result { - let MemtableRanges { ranges } = mem_ranges; + let components = group_by_sequence( + mem_ranges.ranges.into_values().collect(), + |range: &MemtableRange| { + let stats = range.stats(); + (stats.min_sequence, stats.max_sequence()) + }, + ); let mut flat_sources = FlatSources { sources: SmallVec::new(), encoded: SmallVec::new(), }; - - if ranges.len() == 1 { - debug!("Flushing single flat range"); - - let only_range = ranges.into_values().next().unwrap(); - let max_sequence = only_range.stats().max_sequence(); - if let Some(encoded) = only_range.encoded() { - flat_sources.encoded.push((encoded, max_sequence)); - } else { - let schema = only_range.record_batch_schema_hint().unwrap_or(schema); - let iter = only_range.build_record_batch_iter(None, None)?; - // Dedup according to append mode and merge mode. - // Even single range may have duplicate rows. - let iter = maybe_dedup_one( - options.append_mode, - options.merge_mode(), - field_column_start, - iter, - ); - flat_sources - .sources - .push((FlatSource::new_iter(schema, iter), max_sequence)); - }; - } else { - let min_flush_rows = *ENCODE_ROW_THRESHOLD; - // Calculate total rows from non-encoded ranges. - let total_rows: usize = ranges - .values() - .filter(|r| r.encoded().is_none()) - .map(|r| r.num_rows()) - .sum(); - debug!( - "Flushing multiple flat ranges, total_rows: {}, min_flush_rows: {}, num_ranges: {}", - total_rows, - min_flush_rows, - ranges.len() - ); - let mut rows_remaining = total_rows; - let mut last_iter_rows = 0; - let num_ranges = ranges.len(); - let mut input_iters = Vec::with_capacity(num_ranges); - let mut current_ranges = Vec::new(); - - let schemas = ranges - .values() - .filter(|range| range.encoded().is_none()) - .map(|range| { - ( - range - .record_batch_schema_hint() - .unwrap_or_else(|| schema.clone()), - range.num_rows() as u64, - ) - }) - .collect::>(); - let plans = Arc::new(collect_json2_rewrite_plans(metadata, &schemas)?); - let schema = rewrite_json2_schema( - schemas.first().map(|(schema, _)| schema).unwrap_or(&schema), - &plans, - ); - - for (_range_id, range) in ranges { - if let Some(encoded) = range.encoded() { - let max_sequence = range.stats().max_sequence(); - flat_sources.encoded.push((encoded, max_sequence)); - continue; - } - - if let Some(actual) = range.record_batch_schema_hint() { - let actual = rewrite_json2_schema(&actual, &plans); - ensure!( - actual == schema, - UnexpectedSnafu { - reason: format!( - "Different schemas found in a MemtableRanges, expected: {}, actual: {}", - schema, actual, - ), - } - ) - } - - let iter = range.build_record_batch_iter(None, None)?; - let iter: BoxedRecordBatchIterator = if plans.is_empty() { - iter - } else { - let plans = plans.clone(); - Box::new(iter.map(move |batch| rewrite_json2_batch(batch?, &plans))) - }; - input_iters.push(iter); - let range_rows = range.num_rows(); - last_iter_rows += range_rows; - rows_remaining -= range_rows; - current_ranges.push(range); - - // Flush if we have enough rows, but don't flush if the remaining rows - // would be less than DEFAULT_ROW_GROUP_SIZE (to avoid small last files). - if last_iter_rows >= min_flush_rows - && (rows_remaining == 0 || rows_remaining >= DEFAULT_ROW_GROUP_SIZE) - { - debug!( - "Flush batch ready, rows: {}, min_rows: {}, num_iters: {}, remaining: {}", - last_iter_rows, - min_flush_rows, - input_iters.len(), - rows_remaining - ); - - // Calculate max_sequence from all merged ranges - let max_sequence = current_ranges - .iter() - .map(|r| r.stats().max_sequence()) - .max() - .unwrap_or(0); - let batch_size = - crate::batch_size::estimate_batch_size(current_ranges.iter().map(|range| { - let stats = range.stats(); - (stats.num_rows() as u64, stats.bytes_allocated() as u64) - })); - - let input_iters = - std::mem::replace(&mut input_iters, Vec::with_capacity(num_ranges)); - - let maybe_dedup = merge_and_dedup_with_batch_size( + let mut rows_remaining = components + .iter() + .filter(|component| component.len() != 1 || component[0].encoded().is_none()) + .flatten() + .map(MemtableRange::num_rows) + .sum::(); + let mut pending = Vec::new(); + let mut pending_rows = 0; + for component in components { + // A standalone encoded component is an ordering barrier. Do not merge + // raw ranges on either side while uploading the intervening rows as-is. + if component.len() == 1 + && let Some(encoded) = component[0].encoded() + { + if !pending.is_empty() { + flat_sources.sources.push(build_flush_source( &schema, - options.append_mode, - options.merge_mode(), + std::mem::take(&mut pending), + metadata, + options, field_column_start, - input_iters, - batch_size, - )?; - - flat_sources.sources.push(( - FlatSource::new_iter(schema.clone(), maybe_dedup), - max_sequence, - )); - last_iter_rows = 0; - current_ranges.clear(); + )?); + pending_rows = 0; } - } - - // Handle remaining iters. - if !input_iters.is_empty() { - debug!( - "Flush remaining batch, rows: {}, min_rows: {}, num_iters: {}, remaining: {}", - last_iter_rows, - min_flush_rows, - input_iters.len(), - rows_remaining - ); - - let max_sequence = current_ranges - .iter() - .map(|r| r.stats().max_sequence()) - .max() - .unwrap_or(0); - let batch_size = - crate::batch_size::estimate_batch_size(current_ranges.iter().map(|range| { - let stats = range.stats(); - (stats.num_rows() as u64, stats.bytes_allocated() as u64) - })); - - let maybe_dedup = merge_and_dedup_with_batch_size( - &schema, - options.append_mode, - options.merge_mode(), - field_column_start, - input_iters, - batch_size, - )?; - flat_sources - .sources - .push((FlatSource::new_iter(schema, maybe_dedup), max_sequence)); + .encoded + .push((encoded, component[0].stats().max_sequence())); + continue; + } + let component_rows = component.iter().map(MemtableRange::num_rows).sum::(); + pending_rows += component_rows; + rows_remaining -= component_rows; + pending.extend(component); + // Only split between disjoint components, retaining the existing + // minimum-size rule for the final source. + if pending_rows >= *ENCODE_ROW_THRESHOLD + && (rows_remaining == 0 || rows_remaining >= DEFAULT_ROW_GROUP_SIZE) + { + flat_sources.sources.push(build_flush_source( + &schema, + std::mem::take(&mut pending), + metadata, + options, + field_column_start, + )?); + pending_rows = 0; } } - + if !pending.is_empty() { + flat_sources.sources.push(build_flush_source( + &schema, + pending, + metadata, + options, + field_column_start, + )?); + } Ok(flat_sources) } +/// Builds one sequence-contiguous source, decoding encoded ranges if their +/// intervals overlap other ranges. JSON2 layouts are reconciled for this group. +fn build_flush_source( + schema: &SchemaRef, + ranges: Vec, + metadata: &RegionMetadataRef, + options: &RegionOptions, + field_column_start: usize, +) -> Result<(FlatSource, SequenceNumber)> { + let max_sequence = ranges + .iter() + .map(|range| range.stats().max_sequence()) + .max() + .unwrap_or(0); + if let [range] = ranges.as_slice() { + // A single range needs no layout reconciliation, including legacy JSON2. + let schema = range + .record_batch_schema_hint() + .unwrap_or_else(|| schema.clone()); + let iter = maybe_dedup_one( + options.append_mode, + options.merge_mode(), + field_column_start, + range.build_record_batch_iter(None, None)?, + ); + return Ok((FlatSource::new_iter(schema, iter), max_sequence)); + } + let batch_size = crate::batch_size::estimate_batch_size(ranges.iter().map(|range| { + let stats = range.stats(); + (stats.num_rows() as u64, stats.bytes_allocated() as u64) + })); + let schemas = ranges + .iter() + .map(|range| { + ( + range + .record_batch_schema_hint() + .unwrap_or_else(|| schema.clone()), + range.num_rows() as u64, + ) + }) + .collect::>(); + let plans = Arc::new(collect_json2_rewrite_plans(metadata, &schemas)?); + let schema = rewrite_json2_schema( + schemas.first().map(|(schema, _)| schema).unwrap_or(schema), + &plans, + ); + let mut iters = Vec::with_capacity(ranges.len()); + for range in ranges { + if let Some(actual) = range.record_batch_schema_hint() { + let actual = rewrite_json2_schema(&actual, &plans); + ensure!( + actual == schema, + UnexpectedSnafu { + reason: format!( + "Different schemas found in a MemtableRanges, expected: {}, actual: {}", + schema, actual + ), + } + ); + } + let iter = range.build_record_batch_iter(None, None)?; + let iter: BoxedRecordBatchIterator = if plans.is_empty() { + iter + } else { + let plans = plans.clone(); + Box::new(iter.map(move |batch| rewrite_json2_batch(batch?, &plans))) + }; + iters.push(iter); + } + let iter = merge_and_dedup_with_batch_size( + &schema, + options.append_mode, + options.merge_mode(), + field_column_start, + iters, + batch_size, + )?; + Ok((FlatSource::new_iter(schema, iter), max_sequence)) +} + /// Merges multiple record batch iterators and applies deduplication based on the specified mode. /// /// This function is used during the flush process to combine data from multiple memtable ranges @@ -1131,7 +1105,7 @@ fn memtable_flat_sources( /// 3. If `append_mode` is false, wraps the merge iterator with a `FlatDedupIterator` that /// applies the specified merge mode: /// - `LastRow`: Removes duplicate rows, keeping only the last one -/// - `LastNonNull`: Removes duplicates but preserves the last non-null value for each field +/// - `LastNonNull`: Folds fields between delete markers, retaining every marker /// /// # Examples /// @@ -2358,7 +2332,7 @@ mod tests { 1, std::iter::once(1000), std::iter::once(Some(1.0)), - 1, + 2, ); converter.append_key_values(&kvs)?; let batch = converter.convert()?.batch; @@ -2373,6 +2347,8 @@ mod tests { )), MemtableStats { num_rows: 1, + min_sequence: id as u64 + 1, + max_sequence: id as u64 + 1, ..Default::default() }, ) @@ -2391,7 +2367,7 @@ mod tests { ranges.insert( 1, new_range( - 0, + 1, TestIterBuilder { schema: schema.clone(), batch: Some(batch), diff --git a/src/mito2/src/memtable.rs b/src/mito2/src/memtable.rs index f6481fbcdce..f3bcd6d08f8 100644 --- a/src/mito2/src/memtable.rs +++ b/src/mito2/src/memtable.rs @@ -64,6 +64,32 @@ pub use bulk::part::{ #[cfg(any(test, feature = "test"))] pub use time_partition::filter_record_batch; +/// Groups sequence intervals into ordered, disjoint components. A component must +/// not be split during a partial merge: an excluded interval could contain an +/// intermediate row version. Bounds may conservatively include removed rows. +pub(crate) fn group_by_sequence( + mut parts: Vec, + bounds: impl Fn(&T) -> (SequenceNumber, SequenceNumber), +) -> Vec> { + parts.sort_by_key(&bounds); + let mut groups: Vec> = Vec::new(); + let mut group_max = 0; + for part in parts { + let (min, max) = bounds(&part); + match groups.last_mut() { + Some(group) if min <= group_max => { + group.push(part); + group_max = group_max.max(max); + } + _ => { + groups.push(vec![part]); + group_max = max; + } + } + } + groups +} + /// Id for memtables. /// /// Should be unique under the same region. @@ -150,6 +176,10 @@ pub struct MemtableStats { pub num_ranges: usize, /// The maximum sequence number in the memtable. pub max_sequence: SequenceNumber, + /// Lower bound of the row sequences written to this memtable or range. + /// May remain below the actual minimum after deduplication. Zero is conservative + /// when a range does not track its minimum; the value is unused for empty memtables. + pub min_sequence: SequenceNumber, /// Number of estimated timeseries in memtable. pub series_count: usize, } @@ -283,6 +313,11 @@ pub trait Memtable: Send + Sync + fmt::Debug { /// Returns the [MemtableStats] info of Memtable. fn stats(&self) -> MemtableStats; + /// Returns a conservative row-sequence lower bound without computing full statistics. + /// Returns zero when write statistics are unavailable. Callers must ignore + /// empty memtables when using this bound to constrain compaction. + fn min_sequence(&self) -> SequenceNumber; + /// Forks this (immutable) memtable and returns a new mutable memtable with specific memtable `id`. /// /// A region must freeze the memtable before invoking this method. diff --git a/src/mito2/src/memtable/bulk.rs b/src/mito2/src/memtable/bulk.rs index 55d8f71e9b0..8533401328a 100644 --- a/src/mito2/src/memtable/bulk.rs +++ b/src/mito2/src/memtable/bulk.rs @@ -21,7 +21,7 @@ pub mod part_reader; mod row_group_reader; use std::collections::{BTreeMap, HashSet}; -use std::sync::atomic::{AtomicI64, AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, LazyLock, Mutex, RwLock}; use std::time::Instant; @@ -40,7 +40,7 @@ use rayon::prelude::*; use serde::{Deserialize, Serialize}; use serde_with::{DisplayFromStr, serde_as}; use store_api::metadata::RegionMetadataRef; -use store_api::storage::{ColumnId, FileId, RegionId, SequenceRange}; +use store_api::storage::{ColumnId, FileId, RegionId, SequenceNumber, SequenceRange}; use tokio::sync::Semaphore; use crate::compaction::{ @@ -59,6 +59,7 @@ use crate::memtable::{ AllocTracker, BoxedBatchIterator, BoxedRecordBatchIterator, EncodedBulkPart, EncodedRange, IterBuilder, KeyValues, MemScanMetrics, Memtable, MemtableBuilder, MemtableId, MemtableRange, MemtableRangeContext, MemtableRanges, MemtableRef, MemtableStats, RangesOptions, + group_by_sequence, }; use crate::read::flat_dedup::{FlatDedupIterator, FlatLastNonNull, FlatLastRow}; use crate::read::flat_merge::FlatMergeIterator; @@ -176,7 +177,7 @@ enum MergedPart { /// Result of collecting parts to merge struct CollectedParts { - /// Groups of parts ready for merging (each group has up to 16 parts) + /// Consecutive groups of parts ready for merging. groups: Vec>, } @@ -201,38 +202,18 @@ impl BulkParts { self.unordered_part.is_empty() && self.parts.is_empty() } - /// Returns true if enough parts of the same type are available to merge. - fn should_merge_parts(&self, merge_threshold: usize, encode_bytes_threshold: usize) -> bool { - let mut bulk_count = 0; - let mut encoded_count = 0; - - for wrapper in &self.parts { - if !Self::is_merge_candidate(wrapper, encode_bytes_threshold) { - continue; - } - - if wrapper.part.is_encoded() { - encoded_count += 1; - } else { - bulk_count += 1; - } - - if bulk_count >= merge_threshold || encoded_count >= merge_threshold { - return true; - } - } - - false - } - - /// Returns whether a part is small enough to benefit from merging. - fn is_merge_candidate(wrapper: &BulkPartWrapper, encode_bytes_threshold: usize) -> bool { - !wrapper.merging - && Self::is_merge_candidate_by_size( - wrapper.part.is_encoded(), - wrapper.part.estimated_size(), - encode_bytes_threshold, - ) + /// Returns a conservative scheduling hint without sorting or allocating. + /// The picker checks sequence barriers and sizes when compaction runs. + fn may_merge_parts(&self, merge_threshold: usize) -> bool { + let merge_threshold = merge_threshold.max(2); + self.parts.len() >= merge_threshold + && self + .parts + .iter() + .filter(|wrapper| !wrapper.merging) + .take(merge_threshold) + .count() + == merge_threshold } fn is_merge_candidate_by_size( @@ -249,79 +230,105 @@ impl BulkParts { || self.unordered_part.estimated_bytes() > encode_bytes_threshold } - /// Collects unmerged parts and marks them as being merged. - /// Only collects parts of types that meet the threshold. - /// Parts are grouped by row count for parallel processing. + /// Converts buffered small writes into a sorted part once a threshold is reached. + fn seal_unordered(&mut self, metadata: &RegionMetadataRef) -> Result<()> { + if let Some(part) = self.unordered_part.to_bulk_part(metadata)? { + self.parts.push(BulkPartWrapper { + part: PartToMerge::Bulk { + part, + file_id: FileId::random(), + }, + merging: false, + }); + self.unordered_part.clear(); + } + Ok(()) + } + + /// Selects consecutive sequence components without crossing unordered, busy, + /// or oversized parts. Overlapping components are indivisible, even above the size limit. + fn pick_merge_groups( + &self, + merge_threshold: usize, + max_merge_groups: usize, + encode_bytes_threshold: usize, + ) -> Vec> { + if self.parts.len() < merge_threshold.max(2) || max_merge_groups == 0 { + return Vec::new(); + } + // The unordered interval participates in overlap grouping as a barrier, + // so neither overlapping parts nor parts on opposite sides can be merged. + let intervals = self + .parts + .iter() + .enumerate() + .map(|(index, wrapper)| { + let part = &wrapper.part; + (Some(index), (part.min_sequence(), part.max_sequence())) + }) + .chain( + self.unordered_part + .sequence_bounds() + .map(|bounds| (None, bounds)), + ) + .collect(); + let components = group_by_sequence(intervals, |(_, bounds)| *bounds); + let mut groups = Vec::new(); + let mut pending = Vec::new(); + for component in components { + if groups.len() >= max_merge_groups { + break; + } + let Some(component) = component + .into_iter() + .map(|(index, _)| index) + .collect::>>() + else { + pending.clear(); + continue; + }; + let busy = component.iter().any(|&i| self.parts[i].merging); + let oversized = component.len() == 1 && { + let part = &self.parts[component[0]].part; + !Self::is_merge_candidate_by_size( + part.is_encoded(), + part.estimated_size(), + encode_bytes_threshold, + ) + }; + if busy || oversized { + pending.clear(); + continue; + } + pending.extend(component); + if pending.len() >= merge_threshold.max(2) { + groups.push(std::mem::take(&mut pending)); + } + } + groups + } + fn collect_parts_to_merge( &mut self, merge_threshold: usize, max_merge_groups: usize, encode_bytes_threshold: usize, ) -> CollectedParts { - let mut bulk_indices = Vec::new(); - let mut encoded_indices = Vec::new(); - - for (idx, wrapper) in self.parts.iter().enumerate() { - if !Self::is_merge_candidate(wrapper, encode_bytes_threshold) { - continue; - } - let num_rows = wrapper.part.num_rows(); - if wrapper.part.is_encoded() { - encoded_indices.push((idx, num_rows)); - } else { - bulk_indices.push((idx, num_rows)); - } - } - - let mut groups = Vec::new(); - - // Process bulk parts if threshold met - if bulk_indices.len() >= merge_threshold { - groups.extend(self.collect_and_group_parts( - bulk_indices, - merge_threshold, - max_merge_groups, - )); - } - - // Process encoded parts if threshold met - if encoded_indices.len() >= merge_threshold { - groups.extend(self.collect_and_group_parts( - encoded_indices, - merge_threshold, - max_merge_groups, - )); - } - - CollectedParts { groups } - } - - /// Sorts indices by row count, groups into chunks, marks as merging, and returns groups. - fn collect_and_group_parts( - &mut self, - mut indices: Vec<(usize, usize)>, - merge_threshold: usize, - max_merge_groups: usize, - ) -> Vec> { - if indices.is_empty() { - return Vec::new(); - } - - indices.sort_unstable_by_key(|(_, num_rows)| *num_rows); - indices - .chunks(merge_threshold) - .take(max_merge_groups) - .map(|chunk| { - chunk - .iter() - .map(|(idx, _)| { - let wrapper = &mut self.parts[*idx]; + let groups = self + .pick_merge_groups(merge_threshold, max_merge_groups, encode_bytes_threshold) + .into_iter() + .map(|group| { + group + .into_iter() + .map(|i| { + let wrapper = &mut self.parts[i]; wrapper.merging = true; wrapper.part.clone() }) .collect() }) - .collect() + .collect(); + CollectedParts { groups } } /// Installs merged parts and removes the original parts by file ids. @@ -424,11 +431,14 @@ pub struct BulkMemtable { max_timestamp: AtomicI64, min_timestamp: AtomicI64, max_sequence: AtomicU64, + min_sequence: AtomicU64, num_rows: AtomicUsize, /// Compactor for merging bulk parts compactor: Arc>, /// Dispatcher for scheduling compaction tasks compact_dispatcher: Option>, + /// Whether a write-triggered compaction is queued or running. + compacting: Arc, /// Whether the append mode is enabled append_mode: bool, /// Mode to handle duplicate rows while merging @@ -476,6 +486,7 @@ impl Memtable for BulkMemtable { max_ts: fragment.max_timestamp, num_rows: fragment.num_rows(), max_sequence: fragment.sequence, + min_sequence: fragment.min_sequence, }; { @@ -489,18 +500,8 @@ impl Memtable for BulkMemtable { bulk_parts.unordered_part.push(fragment); // Compacts unordered_part if the row or byte threshold is exceeded. - if bulk_parts.should_compact_unordered_part(self.config.encode_bytes_threshold) - && let Some(bulk_part) = - bulk_parts.unordered_part.to_bulk_part(&self.metadata)? - { - bulk_parts.parts.push(BulkPartWrapper { - part: PartToMerge::Bulk { - part: bulk_part, - file_id: FileId::random(), - }, - merging: false, - }); - bulk_parts.unordered_part.clear(); + if bulk_parts.should_compact_unordered_part(self.config.encode_bytes_threshold) { + bulk_parts.seal_unordered(&self.metadata)?; } } else { bulk_parts.parts.push(BulkPartWrapper { @@ -519,9 +520,7 @@ impl Memtable for BulkMemtable { self.update_stats(local_metrics); } - if self.should_compact() { - self.schedule_compact(); - } + self.schedule_compact(); Ok(()) } @@ -637,6 +636,7 @@ impl Memtable for BulkMemtable { num_rows: 0, num_ranges: 0, max_sequence: 0, + min_sequence: 0, series_count: 0, }; } @@ -660,10 +660,18 @@ impl Memtable for BulkMemtable { num_rows: self.num_rows.load(Ordering::Relaxed), num_ranges, max_sequence: self.max_sequence.load(Ordering::Relaxed), + min_sequence: self.min_sequence.load(Ordering::Relaxed), series_count: self.estimated_series_count(), } } + fn min_sequence(&self) -> SequenceNumber { + if self.alloc_tracker.bytes_allocated() == 0 || self.num_rows.load(Ordering::Relaxed) == 0 { + return 0; + } + self.min_sequence.load(Ordering::Relaxed) + } + fn fork(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef { Arc::new(Self { id, @@ -674,6 +682,7 @@ impl Memtable for BulkMemtable { max_timestamp: AtomicI64::new(i64::MIN), min_timestamp: AtomicI64::new(i64::MAX), max_sequence: AtomicU64::new(0), + min_sequence: AtomicU64::new(u64::MAX), num_rows: AtomicUsize::new(0), compactor: Arc::new(Mutex::new(MemtableCompactor::new( metadata.region_id, @@ -682,6 +691,7 @@ impl Memtable for BulkMemtable { self.row_group_size, ))), compact_dispatcher: self.compact_dispatcher.clone(), + compacting: Arc::new(AtomicBool::new(false)), append_mode: self.append_mode, merge_mode: self.merge_mode, row_group_size: self.row_group_size, @@ -695,21 +705,12 @@ impl Memtable for BulkMemtable { return Ok(()); } - // Unified merge for all parts - let should_merge = self.parts.read().unwrap().should_merge_parts( - self.config.merge_threshold, - self.config.encode_bytes_threshold, - ); - if should_merge { - compactor.merge_parts( - &self.parts, - &self.metadata, - !self.append_mode, - self.merge_mode, - )?; - } - - Ok(()) + compactor.merge_parts( + &self.parts, + &self.metadata, + !self.append_mode, + self.merge_mode, + ) } } @@ -759,6 +760,7 @@ impl BulkMemtable { max_timestamp: AtomicI64::new(i64::MIN), min_timestamp: AtomicI64::new(i64::MAX), max_sequence: AtomicU64::new(0), + min_sequence: AtomicU64::new(u64::MAX), num_rows: AtomicUsize::new(0), compactor: Arc::new(Mutex::new(MemtableCompactor::new( region_id, @@ -767,6 +769,7 @@ impl BulkMemtable { row_group_size, ))), compact_dispatcher, + compacting: Arc::new(AtomicBool::new(false)), append_mode, merge_mode, row_group_size, @@ -806,6 +809,8 @@ impl BulkMemtable { .fetch_min(stats.min_ts, Ordering::Relaxed); self.max_sequence .fetch_max(stats.max_sequence, Ordering::Relaxed); + self.min_sequence + .fetch_min(stats.min_sequence, Ordering::Relaxed); self.num_rows.fetch_add(stats.num_rows, Ordering::Relaxed); } @@ -819,22 +824,27 @@ impl BulkMemtable { .sum() } - /// Returns whether the memtable should be compacted. - fn should_compact(&self) -> bool { - let parts = self.parts.read().unwrap(); - parts.should_merge_parts( - self.config.merge_threshold, - self.config.encode_bytes_threshold, - ) - } - - /// Schedules a compaction task using the CompactDispatcher. + /// Schedules at most one write-triggered compaction at a time. fn schedule_compact(&self) { + if self.config.max_merge_groups == 0 || self.compacting.load(Ordering::Relaxed) { + return; + } + if !self + .parts + .read() + .unwrap() + .may_merge_parts(self.config.merge_threshold) + { + return; + } + let Some(guard) = CompactTaskGuard::try_new(self.compacting.clone()) else { + return; + }; if let Some(dispatcher) = &self.compact_dispatcher { let task = MemCompactTask { metadata: self.metadata.clone(), parts: self.parts.clone(), - config: self.config.clone(), + _guard: guard, compactor: self.compactor.clone(), append_mode: self.append_mode, merge_mode: self.merge_mode, @@ -1059,6 +1069,15 @@ impl PartToMerge { } } + /// Returns the lower bound of input sequences. + fn min_sequence(&self) -> SequenceNumber { + match self { + PartToMerge::Bulk { part, .. } => part.min_sequence, + PartToMerge::Multi { part, .. } => part.min_sequence(), + PartToMerge::Encoded { part, .. } => part.metadata().min_sequence, + } + } + /// Gets the maximum sequence number of this part. fn max_sequence(&self) -> u64 { match self { @@ -1207,12 +1226,14 @@ impl MemtableCompactor { ) -> Result<()> { let start = Instant::now(); - // Collect pre-grouped parts - let collected = bulk_parts.write().unwrap().collect_parts_to_merge( - self.config.merge_threshold, - self.config.max_merge_groups, - self.config.encode_bytes_threshold, - ); + let collected = { + let mut parts = bulk_parts.write().unwrap(); + parts.collect_parts_to_merge( + self.config.merge_threshold, + self.config.max_merge_groups, + self.config.encode_bytes_threshold, + ) + }; if collected.groups.is_empty() { return Ok(()); @@ -1298,6 +1319,11 @@ impl MemtableCompactor { .map(|p| p.max_timestamp()) .max() .unwrap_or(i64::MIN); + let min_sequence = parts_to_merge + .iter() + .map(PartToMerge::min_sequence) + .min() + .unwrap_or(0); let max_sequence = parts_to_merge .iter() .map(|p| p.max_sequence()) @@ -1389,7 +1415,7 @@ impl MemtableCompactor { schema, min_timestamp, max_timestamp, - max_sequence, + min_sequence..=max_sequence, &mut metrics, )?; @@ -1415,6 +1441,7 @@ impl MemtableCompactor { batches, min_timestamp, max_timestamp, + min_sequence, max_sequence, estimated_series_count, metadata, @@ -1431,12 +1458,31 @@ impl MemtableCompactor { } } +/// Keeps the compaction flag set while queued or running and clears it on drop, +/// including failures and cancellation while waiting for a dispatcher permit. +struct CompactTaskGuard(Arc); + +impl CompactTaskGuard { + fn try_new(compacting: Arc) -> Option { + // The compactor mutex protects merging; this flag only avoids duplicate tasks. + compacting + .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed) + .ok() + .map(|_| Self(compacting)) + } +} + +impl Drop for CompactTaskGuard { + fn drop(&mut self) { + self.0.store(false, Ordering::Relaxed); + } +} + /// A memtable compact task to run in background. struct MemCompactTask { metadata: RegionMetadataRef, parts: Arc>, - /// Configuration for the bulk memtable. - config: BulkMemtableConfig, + _guard: CompactTaskGuard, /// Compactor for merging bulk parts compactor: Arc>, /// Whether the append mode is enabled @@ -1447,22 +1493,12 @@ struct MemCompactTask { impl MemCompactTask { fn compact(&self) -> Result<()> { - let mut compactor = self.compactor.lock().unwrap(); - - let should_merge = self.parts.read().unwrap().should_merge_parts( - self.config.merge_threshold, - self.config.encode_bytes_threshold, - ); - if should_merge { - compactor.merge_parts( - &self.parts, - &self.metadata, - !self.append_mode, - self.merge_mode, - )?; - } - - Ok(()) + self.compactor.lock().unwrap().merge_parts( + &self.parts, + &self.metadata, + !self.append_mode, + self.merge_mode, + ) } } @@ -1591,7 +1627,8 @@ mod tests { use api::v1::value::ValueData; use api::v1::{Mutation, Row, Rows, SemanticType}; use common_error::ext::WhateverResult; - use datatypes::arrow::datatypes::DataType as ArrowDataType; + use datatypes::arrow::array::AsArray; + use datatypes::arrow::datatypes::{DataType as ArrowDataType, UInt64Type}; use datatypes::data_type::ConcreteDataType; use datatypes::extension::json::{ JSON2_REMAINDER_FIELD_NAME, Json2ExtensionType, Json2PhysicalLayout, JsonMetadata, @@ -1607,6 +1644,7 @@ mod tests { use super::*; use crate::memtable::bulk::part::BulkPartConverter; use crate::read::scan_region::PredicateGroup; + use crate::sst::parquet::flat_format::sequence_column_index; use crate::sst::{FlatSchemaOptions, to_flat_sst_arrow_schema}; use crate::test_util::memtable_util::{ build_key_values_with_ts_seq_values, metadata_for_test, region_metadata_to_row_schema, @@ -1680,6 +1718,69 @@ mod tests { ); } + #[test] + fn test_min_sequence_survives_mixed_parts_compaction_and_fork() { + let metadata = metadata_for_test(); + let config = BulkMemtableConfig { + merge_threshold: 3, + encode_row_threshold: 1, + encode_bytes_threshold: 1, + ..Default::default() + }; + let memtable = BulkMemtable::new( + 1, + config, + metadata.clone(), + None, + None, + false, + MergeMode::LastNonNull, + ); + memtable.set_unordered_part_threshold(0); + for (sequence, expected_min, expected_ranges) in [(100, 100, 1), (10, 10, 2), (200, 10, 1)] + { + let part = create_bulk_part_with_converter( + "a", + 1, + vec![2000, 1000], + vec![Some(1.0), Some(2.0)], + sequence, + ) + .unwrap(); + // Regular row writes produce heterogeneous parts, sorted by timestamp + // rather than sequence. Neither the scalar nor the first row is a minimum. + assert_eq!(sequence + 1, part.sequence); + assert_eq!(sequence, part.min_sequence); + let sequences = part + .batch + .column(sequence_column_index(part.batch.num_columns())) + .as_primitive::(); + assert_eq!(&[sequence + 1, sequence], sequences.values().as_ref()); + memtable.write_bulk(part).unwrap(); + assert_eq!(expected_min, memtable.stats().min_sequence); + assert_eq!(expected_min, memtable.min_sequence()); + assert_eq!(expected_ranges, memtable.stats().num_ranges); + assert!(!memtable.compacting.load(Ordering::Relaxed)); + } + // The third write synchronously merges all parts without a dispatcher. + // Dedup can remove the oldest rows, but their lower bound remains safe. + assert_eq!(10, memtable.stats().min_sequence); + assert_eq!(201, memtable.stats().max_sequence); + let ranges = memtable.ranges(None, RangesOptions::default()).unwrap(); + let merged = ranges.ranges.values().next().unwrap(); + assert!(merged.encoded().is_some()); + assert_eq!(10, merged.stats().min_sequence); + let fork = memtable.fork(2, &metadata); + assert!(fork.is_empty()); + assert_eq!(0, fork.min_sequence()); + fork.write_bulk( + create_bulk_part_with_converter("a", 1, vec![1000], vec![Some(1.0)], 50).unwrap(), + ) + .unwrap(); + assert_eq!(50, fork.stats().min_sequence); + assert_eq!(50, fork.min_sequence()); + } + #[test] fn test_bulk_memtable_write_read() { let metadata = metadata_for_test(); @@ -1724,6 +1825,7 @@ mod tests { assert_eq!(5, stats.num_rows); assert_eq!(3, stats.num_ranges); assert_eq!(300, stats.max_sequence); + assert_eq!(100, stats.min_sequence); let (min_ts, max_ts) = stats.time_range.unwrap(); assert_eq!(1000, min_ts.value()); @@ -2458,7 +2560,7 @@ mod tests { assert_eq!(5, large_part.num_rows()); memtable.write_bulk(large_part).unwrap(); - // Write another small part (2 rows) - should trigger compaction of unordered_part + // A later small write fills the unordered part to its compact threshold. let part = create_bulk_part_with_converter( "small_2", 2, @@ -2623,10 +2725,10 @@ mod tests { } #[test] - fn test_should_merge_parts_below_threshold() { + fn test_may_merge_parts_below_threshold() { let mut bulk_parts = BulkParts::default(); - // Add 7 bulk parts (below DEFAULT_MERGE_THRESHOLD of 8) + // Add one fewer part than the merge threshold. for i in 0..DEFAULT_MERGE_THRESHOLD - 1 { let part = create_bulk_part_with_converter( &format!("key_{}", i), @@ -2637,14 +2739,18 @@ mod tests { ) .unwrap(); bulk_parts.parts.push(create_bulk_part_wrapper(part)); + if i == 0 { + assert!(!bulk_parts.may_merge_parts(0)); + assert!(!bulk_parts.may_merge_parts(1)); + } } - // Should not trigger merge since we have only 7 parts - assert!(!bulk_parts.should_merge_parts(DEFAULT_MERGE_THRESHOLD, usize::MAX)); + // There are not enough parts to merge. + assert!(!bulk_parts.may_merge_parts(DEFAULT_MERGE_THRESHOLD)); } #[test] - fn test_should_merge_parts_at_threshold() { + fn test_may_merge_parts_at_threshold() { let mut bulk_parts = BulkParts::default(); let merge_threshold = 8; @@ -2662,7 +2768,7 @@ mod tests { } // Should trigger merge since we have 8 parts - assert!(bulk_parts.should_merge_parts(merge_threshold, usize::MAX)); + assert!(bulk_parts.may_merge_parts(merge_threshold)); } #[test] @@ -2682,7 +2788,7 @@ mod tests { } let max_size = bulk_parts.parts[0].part.estimated_size(); - assert!(bulk_parts.should_merge_parts(merge_threshold, max_size)); + assert!(bulk_parts.may_merge_parts(merge_threshold)); assert_eq!( 1, bulk_parts @@ -2693,7 +2799,7 @@ mod tests { } #[test] - fn test_should_merge_parts_with_merging_flag() { + fn test_may_merge_parts_with_merging_flag() { let mut bulk_parts = BulkParts::default(); let merge_threshold = 8; @@ -2711,7 +2817,7 @@ mod tests { } // Should trigger merge since we have 10 parts - assert!(bulk_parts.should_merge_parts(merge_threshold, usize::MAX)); + assert!(bulk_parts.may_merge_parts(merge_threshold)); // Mark first 3 parts as merging for wrapper in bulk_parts.parts.iter_mut().take(3) { @@ -2719,53 +2825,115 @@ mod tests { } // Now only 7 parts are available for merging, should not trigger - assert!(!bulk_parts.should_merge_parts(merge_threshold, usize::MAX)); + assert!(!bulk_parts.may_merge_parts(merge_threshold)); } #[test] - fn test_collect_parts_to_merge_grouping() { - let mut bulk_parts = BulkParts::default(); + fn test_compact_task_guard() { + let compacting = Arc::new(AtomicBool::new(false)); + let guard = CompactTaskGuard::try_new(compacting.clone()).unwrap(); + assert!(compacting.load(Ordering::Relaxed)); + assert!(CompactTaskGuard::try_new(compacting.clone()).is_none()); + drop(guard); + assert!(!compacting.load(Ordering::Relaxed)); + assert!(CompactTaskGuard::try_new(compacting.clone()).is_some()); + assert!(!compacting.load(Ordering::Relaxed)); + } - // Add 16 bulk parts with different row counts - for i in 0..16 { - let num_rows = (i % 4) + 1; // 1 to 4 rows - let timestamps: Vec = (0..num_rows) - .map(|j| 1000 + i as i64 * 100 + j as i64) - .collect(); - let values: Vec> = - (0..num_rows).map(|j| Some((i * 10 + j) as f64)).collect(); - let part = create_bulk_part_with_converter( - &format!("key_{}", i), - i as u32, - timestamps, - values, - 100 + i as u64, - ) - .unwrap(); - bulk_parts.parts.push(create_bulk_part_wrapper(part)); - } - - // Should trigger merge since we have 16 parts - assert!(bulk_parts.should_merge_parts(DEFAULT_MERGE_THRESHOLD, usize::MAX)); - - // Collect parts to merge - let collected = bulk_parts.collect_parts_to_merge( - DEFAULT_MERGE_THRESHOLD, - DEFAULT_MAX_MERGE_GROUPS, - usize::MAX, + #[test] + fn test_compact_scheduling_disabled() { + let memtable = BulkMemtable::new( + 0, + BulkMemtableConfig { + merge_threshold: 2, + max_merge_groups: 0, + ..Default::default() + }, + metadata_for_test(), + None, + Some(Arc::new(CompactDispatcher::new(0))), + false, + MergeMode::LastRow, ); + let part = + create_bulk_part_with_converter("key", 0, vec![1000], vec![Some(1.0)], 0).unwrap(); + memtable.parts.write().unwrap().parts = vec![ + create_bulk_part_wrapper(part.clone()), + create_bulk_part_wrapper(part), + ]; + memtable.schedule_compact(); + assert!(!memtable.compacting.load(Ordering::Relaxed)); + } - // Should have groups - assert!(!collected.groups.is_empty()); + #[test] + fn test_collect_parts_to_merge_sequence_components() { + let make_part = |min, max| { + let mut part = + create_bulk_part_with_converter("key", 0, vec![1000], vec![Some(1.0)], max) + .unwrap(); + part.min_sequence = min; + part + }; + let metadata = metadata_for_test(); + let encoder = BulkPartEncoder::new(metadata.clone(), DEFAULT_ROW_GROUP_SIZE).unwrap(); + // Deliberately use installation order rather than sequence order. + let mut parts = BulkParts { + parts: vec![ + create_bulk_part_wrapper(make_part(70, 70)), + create_bulk_part_wrapper(make_part(10, 30)), + BulkPartWrapper { + part: PartToMerge::Multi { + part: MultiBulkPart::from_bulk_part(make_part(35, 50), &metadata), + file_id: FileId::random(), + }, + merging: false, + }, + create_bulk_part_wrapper(make_part(60, 60)), + BulkPartWrapper { + part: PartToMerge::Encoded { + part: encoder.encode_part(&make_part(20, 40)).unwrap().unwrap(), + file_id: FileId::random(), + }, + merging: false, + }, + ], + ..Default::default() + }; + // The oversized encoded part connects three intervals transitively; + // neither the merge threshold nor its size may split that component. + assert!(parts.may_merge_parts(parts.parts.len())); + assert_eq!( + vec![vec![1, 4, 2], vec![3, 0]], + parts.pick_merge_groups(2, 2, 1) + ); + // The hint may schedule a merge even when unordered writes block all parts. + parts.unordered_part.push(make_part(0, 100)); + assert!(parts.may_merge_parts(2)); + assert!(parts.pick_merge_groups(2, 2, 1).is_empty()); + parts.unordered_part.clear(); - // All groups should have parts - for group in &collected.groups { - assert!(!group.is_empty()); - } + // Touching the unordered interval blocks the entire transitive component. + parts.unordered_part.push(make_part(50, 55)); + assert_eq!(vec![vec![3, 0]], parts.pick_merge_groups(2, 2, 1)); + parts.unordered_part.clear(); - // Total parts collected should be 16 - let total_parts: usize = collected.groups.iter().map(|g| g.len()).sum(); - assert_eq!(16, total_parts); + // Even without overlap, a group cannot straddle pending small writes. + parts.unordered_part.push(make_part(65, 65)); + assert_eq!(vec![vec![1, 4, 2]], parts.pick_merge_groups(2, 2, 1)); + parts.unordered_part.clear(); + + parts.parts[4].merging = true; + let selected = parts.collect_parts_to_merge(2, 2, 1); + assert_eq!(1, selected.groups.len()); + assert_eq!( + vec![60, 70], + selected.groups[0] + .iter() + .map(PartToMerge::max_sequence) + .collect::>() + ); + assert!(!parts.parts[1].merging); + assert!(!parts.parts[2].merging); } #[test] diff --git a/src/mito2/src/memtable/bulk/part.rs b/src/mito2/src/memtable/bulk/part.rs index 2a0c42b7035..9f1051f3282 100644 --- a/src/mito2/src/memtable/bulk/part.rs +++ b/src/mito2/src/memtable/bulk/part.rs @@ -15,6 +15,7 @@ //! Bulk part encoder/decoder. use std::collections::{HashMap, HashSet}; +use std::ops::RangeInclusive; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -80,7 +81,12 @@ pub struct BulkPart { pub batch: RecordBatch, pub max_timestamp: i64, pub min_timestamp: i64, + /// Uniform sequence for external bulk writes, or the maximum row sequence + /// for parts produced by [`BulkPartConverter`]. Rows need not share it. pub sequence: u64, + /// Lower bound of row sequences. Slices may inherit a lower bound from + /// their parent part rather than recomputing an exact minimum. + pub min_sequence: SequenceNumber, pub timestamp_index: usize, pub raw_data: Option, } @@ -101,6 +107,9 @@ impl TryFrom for BulkPart { max_timestamp: value.max_ts, min_timestamp: value.min_ts, sequence: value.sequence, + // Bulk WAL entries represent uniform-sequence external writes. + // Row mutations reconstruct their bounds through BulkPartConverter. + min_sequence: value.sequence, timestamp_index: value.timestamp_index as usize, raw_data: Some(ipc), }) @@ -183,6 +192,7 @@ impl BulkPart { num_rows: self.num_rows(), num_ranges: 1, max_sequence: self.sequence, + min_sequence: self.min_sequence, series_count: self.estimated_series_count(), } } @@ -353,6 +363,8 @@ pub struct UnorderedPart { max_timestamp: i64, /// Maximum sequence number across all parts. max_sequence: u64, + /// Minimum sequence lower bound across all parts. + min_sequence: SequenceNumber, /// Row count threshold for accepting parts (default: 1024). threshold: usize, /// Row count threshold for compacting (default: 4096). @@ -375,6 +387,7 @@ impl UnorderedPart { min_timestamp: i64::MAX, max_timestamp: i64::MIN, max_sequence: 0, + min_sequence: SequenceNumber::MAX, threshold: 1024, compact_threshold: 4096, } @@ -415,6 +428,11 @@ impl UnorderedPart { self.total_bytes } + /// Returns the sequence interval occupied by pending small writes. + pub(super) fn sequence_bounds(&self) -> Option<(SequenceNumber, SequenceNumber)> { + (!self.is_empty()).then_some((self.min_sequence, self.max_sequence)) + } + /// Adds a BulkPart to this unordered collection. pub fn push(&mut self, part: BulkPart) { self.total_rows += part.num_rows(); @@ -422,6 +440,7 @@ impl UnorderedPart { self.min_timestamp = self.min_timestamp.min(part.min_timestamp); self.max_timestamp = self.max_timestamp.max(part.max_timestamp); self.max_sequence = self.max_sequence.max(part.sequence); + self.min_sequence = self.min_sequence.min(part.min_sequence); self.parts.push(part); } @@ -495,6 +514,7 @@ impl UnorderedPart { max_timestamp: self.max_timestamp, min_timestamp: self.min_timestamp, sequence: self.max_sequence, + min_sequence: self.min_sequence, timestamp_index, raw_data: None, })) @@ -508,6 +528,7 @@ impl UnorderedPart { self.min_timestamp = i64::MAX; self.max_timestamp = i64::MIN; self.max_sequence = 0; + self.min_sequence = SequenceNumber::MAX; } } @@ -579,6 +600,8 @@ pub struct BulkPartConverter { min_ts: i64, /// Max sequence number. max_sequence: SequenceNumber, + /// Min sequence number. + min_sequence: SequenceNumber, } impl BulkPartConverter { @@ -616,6 +639,7 @@ impl BulkPartConverter { min_ts: i64::MAX, max_ts: i64::MIN, max_sequence: SequenceNumber::MIN, + min_sequence: SequenceNumber::MAX, } } @@ -694,6 +718,7 @@ impl BulkPartConverter { self.min_ts = self.min_ts.min(ts); self.max_ts = self.max_ts.max(ts); self.max_sequence = self.max_sequence.max(kv.sequence()); + self.min_sequence = self.min_sequence.min(kv.sequence()); Ok(()) } @@ -734,6 +759,7 @@ impl BulkPartConverter { max_timestamp: self.max_ts, min_timestamp: self.min_ts, sequence: self.max_sequence, + min_sequence: self.min_sequence, timestamp_index, raw_data: None, }) @@ -1141,6 +1167,7 @@ pub fn convert_bulk_part( max_timestamp: part.max_timestamp, min_timestamp: part.min_timestamp, sequence: part.sequence, + min_sequence: part.sequence, timestamp_index: new_timestamp_index, raw_data: None, })) @@ -1194,6 +1221,7 @@ impl EncodedBulkPart { num_rows: meta.num_rows, num_ranges: 1, max_sequence: meta.max_sequence, + min_sequence: meta.min_sequence, series_count: meta.num_series as usize, } } @@ -1280,6 +1308,8 @@ pub struct BulkPartMeta { pub region_metadata: RegionMetadataRef, /// Number of series. pub num_series: u64, + /// Lower bound of input sequences, including rows removed by deduplication. + pub min_sequence: SequenceNumber, /// Maximum sequence number in part. pub max_sequence: u64, } @@ -1338,7 +1368,7 @@ impl BulkPartEncoder { arrow_schema: SchemaRef, min_timestamp: i64, max_timestamp: i64, - max_sequence: u64, + sequence_bounds: RangeInclusive, metrics: &mut BulkPartEncodeMetrics, ) -> Result> { let mut buf = Vec::with_capacity(4096); @@ -1390,7 +1420,8 @@ impl BulkPartEncoder { parquet_metadata, region_metadata: self.metadata.clone(), num_series, - max_sequence, + min_sequence: *sequence_bounds.start(), + max_sequence: *sequence_bounds.end(), }, schema: arrow_schema, })) @@ -1426,6 +1457,7 @@ impl BulkPartEncoder { region_metadata: self.metadata.clone(), num_series: part.estimated_series_count() as u64, max_sequence: part.sequence, + min_sequence: part.min_sequence, }, schema: arrow_schema, })) @@ -1635,6 +1667,8 @@ pub struct MultiBulkPart { max_timestamp: i64, /// Min timestamp in part. min_timestamp: i64, + /// Lower bound of input sequences, including rows removed by deduplication. + min_sequence: SequenceNumber, /// Max sequence number in part. max_sequence: SequenceNumber, /// Number of series. @@ -1658,6 +1692,7 @@ impl MultiBulkPart { total_rows: num_rows, max_timestamp: part.max_timestamp, min_timestamp: part.min_timestamp, + min_sequence: part.min_sequence, max_sequence: part.sequence, series_count, batch_stats, @@ -1670,6 +1705,7 @@ impl MultiBulkPart { /// * `batches` - Ordered record batches /// * `min_timestamp` - Minimum timestamp across all batches /// * `max_timestamp` - Maximum timestamp across all batches + /// * `min_sequence` - Lower bound of input sequences /// * `max_sequence` - Maximum sequence number across all batches /// * `series_count` - Number of series in the batches /// * `metadata` - Region metadata for computing batch statistics @@ -1680,6 +1716,7 @@ impl MultiBulkPart { batches: Vec, min_timestamp: i64, max_timestamp: i64, + min_sequence: SequenceNumber, max_sequence: SequenceNumber, series_count: usize, metadata: &RegionMetadata, @@ -1694,6 +1731,7 @@ impl MultiBulkPart { total_rows, max_timestamp, min_timestamp, + min_sequence, max_sequence, series_count, batch_stats, @@ -1719,6 +1757,11 @@ impl MultiBulkPart { self.max_timestamp } + /// Returns the lower bound of input sequences. + pub fn min_sequence(&self) -> SequenceNumber { + self.min_sequence + } + /// Returns the maximum sequence number. pub fn max_sequence(&self) -> SequenceNumber { self.max_sequence @@ -1809,6 +1852,7 @@ impl MultiBulkPart { num_rows: self.num_rows(), num_ranges: 1, max_sequence: self.max_sequence, + min_sequence: self.min_sequence, series_count: self.series_count, } } @@ -1984,6 +2028,7 @@ mod tests { max_timestamp: 0, min_timestamp: 0, sequence: 0, + min_sequence: 0, timestamp_index: 0, raw_data: None, }; @@ -2692,6 +2737,7 @@ mod tests { max_timestamp: 0, min_timestamp: 0, sequence: 0, + min_sequence: 0, timestamp_index: 0, raw_data: None, }; @@ -2737,6 +2783,7 @@ mod tests { max_timestamp: 2000, min_timestamp: 1000, sequence: 5, + min_sequence: 5, timestamp_index: 4, raw_data: None, }; @@ -2761,6 +2808,7 @@ mod tests { assert_eq!(converted.max_timestamp, 2000); assert_eq!(converted.min_timestamp, 1000); assert_eq!(converted.sequence, 5); + assert_eq!(converted.min_sequence, 5); let schema = converted.batch.schema(); let field_names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect(); @@ -2829,6 +2877,7 @@ mod tests { max_timestamp: 2000, min_timestamp: 1000, sequence: 3, + min_sequence: 3, timestamp_index: 4, raw_data: None, }; @@ -2942,6 +2991,7 @@ mod tests { max_timestamp: 2000, min_timestamp: 1000, sequence: 7, + min_sequence: 7, timestamp_index: 3, raw_data: None, }; @@ -3022,6 +3072,7 @@ mod tests { max_timestamp: 4000, min_timestamp: 1000, sequence: 10, + min_sequence: 10, timestamp_index: 4, raw_data: None, }; @@ -3077,18 +3128,58 @@ mod tests { converter.convert().unwrap() } + #[test] + fn test_min_sequence_survives_slicing_and_unordered_part_reuse() { + let metadata = metadata_for_test(); + let make_part = |sequence| { + build_converted_bulk_part(&[MutationInput { + k0: "a", + k1: 0, + timestamps: &[2000, 1000], + v1: &[Some(1.0), Some(2.0)], + sequence, + }]) + }; + let original = make_part(100); + let sliced = crate::memtable::time_partition::filter_record_batch(&original, 1000, 1001) + .unwrap() + .unwrap(); + // The remaining row has sequence 101. Inheriting 100 is conservative + // and avoids rescanning the sequence column after each partition split. + assert_eq!(1, sliced.num_rows()); + assert_eq!(100, sliced.min_sequence); + assert_eq!(101, sliced.sequence); + + let mut unordered = UnorderedPart::new(); + unordered.push(sliced.clone()); + unordered.push(make_part(10)); + let merged = unordered.to_bulk_part(&metadata).unwrap().unwrap(); + assert_eq!(3, merged.num_rows()); + assert_eq!(10, merged.min_sequence); + assert_eq!(101, merged.sequence); + assert_eq!(10, merged.to_memtable_stats(&metadata).min_sequence); + + unordered.clear(); + unordered.push(sliced); + let reused = unordered.to_bulk_part(&metadata).unwrap().unwrap(); + assert_eq!(1, reused.num_rows()); + assert_eq!(100, reused.min_sequence); + } + /// Helper to create a MultiBulkPart where each group becomes a separate batch. fn build_multi_bulk_part(groups: &[&[MutationInput]]) -> (MultiBulkPart, RegionMetadataRef) { let metadata = metadata_for_test(); let mut all_batches = Vec::new(); let mut min_ts = i64::MAX; let mut max_ts = i64::MIN; + let mut min_seq = u64::MAX; let mut max_seq = 0u64; for inputs in groups { let part = build_converted_bulk_part(inputs); min_ts = min_ts.min(part.min_timestamp); max_ts = max_ts.max(part.max_timestamp); + min_seq = min_seq.min(part.min_sequence); max_seq = max_seq.max(part.sequence); all_batches.push(part.batch); } @@ -3097,6 +3188,7 @@ mod tests { all_batches, min_ts, max_ts, + min_seq, max_seq, groups.len(), &metadata, @@ -3223,6 +3315,7 @@ mod tests { max_timestamp: 2000, min_timestamp: 1000, sequence: 5, + min_sequence: 5, timestamp_index: 3, raw_data: Some(ArrowIpc { schema: schema_bytes, @@ -3241,6 +3334,7 @@ mod tests { // The WAL entry round trip keeps the filled column. let entry = BulkWalEntry::try_from(&part).unwrap(); let replayed = BulkPart::try_from(entry).unwrap(); + assert_eq!(part.sequence, replayed.min_sequence); assert_eq!(2, replayed.num_rows()); assert!(replayed.batch.column_by_name("v1").is_some()); } diff --git a/src/mito2/src/memtable/simple_bulk_memtable.rs b/src/mito2/src/memtable/simple_bulk_memtable.rs index 6ff799ebf5c..89c22f5e69c 100644 --- a/src/mito2/src/memtable/simple_bulk_memtable.rs +++ b/src/mito2/src/memtable/simple_bulk_memtable.rs @@ -27,7 +27,7 @@ use mito_codec::key_values::KeyValue; use rayon::prelude::*; use snafu::{OptionExt, ResultExt}; use store_api::metadata::RegionMetadataRef; -use store_api::storage::ColumnId; +use store_api::storage::{ColumnId, SequenceNumber}; use crate::flush::WriteBufferManagerRef; use crate::memtable::bulk::part::BulkPart; @@ -51,6 +51,7 @@ pub struct SimpleBulkMemtable { max_timestamp: AtomicI64, min_timestamp: AtomicI64, max_sequence: AtomicU64, + min_sequence: AtomicU64, dedup: bool, merge_mode: MergeMode, num_rows: AtomicUsize, @@ -80,6 +81,7 @@ impl SimpleBulkMemtable { max_timestamp: AtomicI64::new(i64::MIN), min_timestamp: AtomicI64::new(i64::MAX), max_sequence: AtomicU64::new(0), + min_sequence: AtomicU64::new(u64::MAX), dedup, merge_mode, num_rows: AtomicUsize::new(0), @@ -114,6 +116,7 @@ impl SimpleBulkMemtable { .value(); stats.min_ts = stats.min_ts.min(ts); stats.max_ts = stats.max_ts.max(ts); + stats.min_sequence = stats.min_sequence.min(sequence); } /// Updates memtable stats. @@ -125,6 +128,8 @@ impl SimpleBulkMemtable { self.min_timestamp.fetch_min(stats.min_ts, Ordering::SeqCst); self.max_sequence .fetch_max(stats.max_sequence, Ordering::SeqCst); + self.min_sequence + .fetch_min(stats.min_sequence, Ordering::SeqCst); } #[cfg(test)] @@ -209,6 +214,7 @@ impl Memtable for SimpleBulkMemtable { max_ts: part.max_timestamp, num_rows: part.num_rows(), max_sequence: sequence, + min_sequence: sequence, }); Ok(()) } @@ -226,6 +232,7 @@ impl Memtable for SimpleBulkMemtable { // Use the memtable's overall time range and max sequence for all ranges let max_sequence = self.max_sequence.load(Ordering::Relaxed); + let min_sequence = self.min_sequence.load(Ordering::Relaxed); let time_range = { let num_rows = self.num_rows.load(Ordering::Relaxed); if num_rows > 0 { @@ -279,6 +286,7 @@ impl Memtable for SimpleBulkMemtable { num_rows, num_ranges: 1, max_sequence, + min_sequence, series_count: 1, }; @@ -329,6 +337,7 @@ impl Memtable for SimpleBulkMemtable { num_rows: 0, num_ranges: 0, max_sequence: 0, + min_sequence: 0, series_count: 0, }; } @@ -341,10 +350,18 @@ impl Memtable for SimpleBulkMemtable { num_rows, num_ranges: 1, max_sequence: self.max_sequence.load(Ordering::Relaxed), + min_sequence: self.min_sequence.load(Ordering::Relaxed), series_count: 1, } } + fn min_sequence(&self) -> SequenceNumber { + if self.num_rows.load(Ordering::Relaxed) == 0 { + return 0; + } + self.min_sequence.load(Ordering::Relaxed) + } + fn fork(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef { Arc::new(Self::new( id, @@ -487,6 +504,29 @@ mod tests { KeyValues::new(metadata, mutation).unwrap() } + #[test] + fn test_min_sequence_covers_single_and_batch_writes_and_fork() { + let memtable = new_test_memtable(true, MergeMode::LastNonNull); + let schema = memtable.schema(); + let rows = [(1, 1.0, "a".to_string()), (2, 2.0, "b".to_string())]; + let newer = build_key_values(schema, 100, &rows, OpType::Put); + assert_eq!(0, memtable.min_sequence()); + memtable.write(&newer).unwrap(); + assert_eq!(100, memtable.stats().min_sequence); + assert_eq!(100, memtable.min_sequence()); + let older = build_key_values(schema, 10, &rows, OpType::Put); + memtable.write_one(older.iter().next().unwrap()).unwrap(); + assert_eq!(10, memtable.stats().min_sequence); + assert_eq!(10, memtable.min_sequence()); + assert_eq!(101, memtable.stats().max_sequence); + let fork = memtable.fork(2, schema); + assert!(fork.is_empty()); + assert_eq!(0, fork.min_sequence()); + fork.write(&newer).unwrap(); + assert_eq!(100, fork.stats().min_sequence); + assert_eq!(100, fork.min_sequence()); + } + #[test] fn test_write_and_iter() { let memtable = new_test_memtable(false, MergeMode::LastRow); @@ -664,6 +704,7 @@ mod tests { let part = BulkPart { batch: rb, sequence: 1, + min_sequence: 1, min_timestamp: 1, max_timestamp: 2, timestamp_index: 0, @@ -851,6 +892,7 @@ mod tests { max_timestamp: 0, min_timestamp: 0, sequence: 0, + min_sequence: 0, timestamp_index: 1, raw_data: None, }) @@ -863,6 +905,7 @@ mod tests { max_timestamp: 1, min_timestamp: 1, sequence: 1, + min_sequence: 1, timestamp_index: 1, raw_data: None, }) diff --git a/src/mito2/src/memtable/stats.rs b/src/mito2/src/memtable/stats.rs index 2058b96bd1b..6a08077feab 100644 --- a/src/mito2/src/memtable/stats.rs +++ b/src/mito2/src/memtable/stats.rs @@ -30,6 +30,8 @@ pub(crate) struct WriteMetrics { pub(crate) num_rows: usize, /// Max sequence number written. pub(crate) max_sequence: SequenceNumber, + /// Min sequence number written. + pub(crate) min_sequence: SequenceNumber, } impl Default for WriteMetrics { @@ -41,6 +43,7 @@ impl Default for WriteMetrics { max_ts: i64::MIN, num_rows: 0, max_sequence: SequenceNumber::MIN, + min_sequence: SequenceNumber::MAX, } } } diff --git a/src/mito2/src/memtable/time_partition.rs b/src/mito2/src/memtable/time_partition.rs index 12d540f9564..a6b27cdd0f3 100644 --- a/src/mito2/src/memtable/time_partition.rs +++ b/src/mito2/src/memtable/time_partition.rs @@ -192,6 +192,7 @@ pub fn filter_record_batch(part: &BulkPart, min: i64, max: i64) -> Result