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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* fix(mito2): include all candidates before last-non-null closure

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* test(mito2): cover deferred non-seed compaction bridges

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <ratuthomm@gmail.com>

* docs(mito2): document compaction overlap index fields

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
(cherry picked from commit 259b42c039)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* 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 <realevenyag@gmail.com>

* fix(mito): respect sequence ranges in bulk merges and flushes

Signed-off-by: evenyag <realevenyag@gmail.com>

* perf(mito): avoid merge selection on the bulk write path

Signed-off-by: evenyag <realevenyag@gmail.com>

---------

Signed-off-by: evenyag <realevenyag@gmail.com>
(cherry picked from commit 674ba72669)
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* test(mito2): match release empty-result rendering in bulk flush test

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
Signed-off-by: evenyag <realevenyag@gmail.com>
Co-authored-by: Yingwen <realevenyag@gmail.com>
This commit is contained in:
Lei, HUANG
2026-10-20 08:31:26 -07:00
committed by GitHub
co-authored by Yingwen
parent 1506974bd9
commit 1e5fef5809
27 changed files with 2579 additions and 532 deletions
@@ -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<BulkPart> {
max_timestamp: max,
min_timestamp: min,
sequence: 0,
min_sequence: 0,
timestamp_index: part.timestamp_index,
raw_data: None,
})
+1 -1
View File
@@ -582,7 +582,7 @@ pub struct SstWriteRequest {
#[allow(dead_code)]
pub storage: Option<String>,
/// 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<SequenceNumber>,
pub sst_write_format: FormatType,
+2
View File
@@ -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;
+15 -2
View File
@@ -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<SequenceNumber>,
/// Inferred compaction time window.
pub(crate) compaction_time_window: Option<Duration>,
}
@@ -83,6 +87,11 @@ impl From<VersionRef> 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,
+610
View File
@@ -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<Option<PickerOutput>> {
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<SequenceNumber>,
) -> 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<CompactionOutput>,
files: Vec<FileHandle>,
metadata: &RegionMetadata,
) -> Vec<CompactionOutput> {
// 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<CompactionOutput>],
seed_by_file: &HashMap<RegionFileId, usize>,
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<FileHandle>,
selected: &mut HashSet<RegionFileId>,
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<FileHandle>) -> CompactionOutput {
CompactionOutput {
output_level: 1,
inputs,
filter_deleted: false,
output_time_range: None,
}
}
fn file_ids(files: &[FileHandle]) -> HashSet<RegionFileId> {
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<u64>,
#[case] busy: bool,
#[case] limit: Option<usize>,
#[case] expected_windows: Vec<Vec<usize>>,
) {
// 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<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, &region.region_options, limit, None);
let picked = picker.pick(&region).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, &region.region_options, limit, None);
let picked = picker.pick(&region).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<u64>,
#[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::<Vec<_>>()
);
}
#[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::<Vec<_>>();
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<u64>,
#[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()),
&region.region_options,
Some(1),
TimestampRange::new(Timestamp::new_second(0), Timestamp::new_second(3600)),
);
let picked = picker.pick(&region).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<u64>,
#[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,
}),
&region.region_options,
Some(1),
TimestampRange::new(Timestamp::new_second(10), Timestamp::new_second(20)),
);
let picked = picker.pick(&region).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()),
&region.region_options,
Some(1),
None,
);
let picked = picker.pick(&region).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(&region).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()));
}
}
+293
View File
@@ -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<FileHandle>,
/// 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<usize>,
/// Active region/file identities mapped to leaf indices in `max_ends` for
/// removal without searching `by_start`. Entries are deleted when visited.
positions: HashMap<RegionFileId, usize>,
/// 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<Option<Timestamp>>,
/// 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<FileHandle>, 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<FileHandle> {
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<usize> {
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<usize>,
query: &OverlapQuery<'_>,
) -> Option<usize> {
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);
}
}
}
}
+17 -8
View File
@@ -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<usize>,
time_range: Option<TimestampRange>,
) -> Arc<dyn Picker> {
@@ -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 &region_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)
}
}
}
@@ -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,
&region_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(),
+45
View File
@@ -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<Item = FileMeta>,
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(),
}
}
+63 -43
View File
@@ -71,6 +71,12 @@ struct WindowPickContext<'a> {
phase: PickPhase,
}
struct WindowOutputContext {
active_window: Option<i64>,
time_window_size: Option<i64>,
max_outputs: Option<usize>,
}
/// 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<i64, Window>,
active_window: Option<i64>,
time_window_size: Option<i64>,
context: WindowOutputContext,
) -> Result<Vec<CompactionOutput>> {
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::<Vec<_>>();
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<Option<PickerOutput>> {
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<usize>,
) -> Result<Option<PickerOutput>> {
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<i64, Window>,
active_window: Option<i64>,
time_window_size: Option<i64>,
) -> Result<Vec<CompactionOutput>> {
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<Item = FileMeta>,
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,
+95 -5
View File
@@ -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::<HashSet<_>>();
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<Option<PickerOutput>> {
// 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!(
+304 -5
View File
@@ -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::<Vec<_>>(),
&[(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::<Vec<_>>(),
&[(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::<Vec<_>>(),
&[(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);
}
+245 -10
View File
@@ -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()
+153 -177
View File
@@ -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<FlatSources> {
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::<Vec<_>>();
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::<usize>();
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::<usize>();
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<MemtableRange>,
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::<Vec<_>>();
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),
+35
View File
@@ -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<T>(
mut parts: Vec<T>,
bounds: impl Fn(&T) -> (SequenceNumber, SequenceNumber),
) -> Vec<Vec<T>> {
parts.sort_by_key(&bounds);
let mut groups: Vec<Vec<T>> = 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.
+381 -213
View File
@@ -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<Vec<PartToMerge>>,
}
@@ -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<Vec<usize>> {
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::<Option<Vec<_>>>()
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<Vec<PartToMerge>> {
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<Mutex<MemtableCompactor>>,
/// Dispatcher for scheduling compaction tasks
compact_dispatcher: Option<Arc<CompactDispatcher>>,
/// Whether a write-triggered compaction is queued or running.
compacting: Arc<AtomicBool>,
/// 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<AtomicBool>);
impl CompactTaskGuard {
fn try_new(compacting: Arc<AtomicBool>) -> Option<Self> {
// 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<RwLock<BulkParts>>,
/// Configuration for the bulk memtable.
config: BulkMemtableConfig,
_guard: CompactTaskGuard,
/// Compactor for merging bulk parts
compactor: Arc<Mutex<MemtableCompactor>>,
/// 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::<UInt64Type>();
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<i64> = (0..num_rows)
.map(|j| 1000 + i as i64 * 100 + j as i64)
.collect();
let values: Vec<Option<f64>> =
(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::<Vec<_>>()
);
assert!(!parts.parts[1].merging);
assert!(!parts.parts[2].merging);
}
#[test]
+96 -2
View File
@@ -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<ArrowIpc>,
}
@@ -101,6 +107,9 @@ impl TryFrom<BulkWalEntry> 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<SequenceNumber>,
metrics: &mut BulkPartEncodeMetrics,
) -> Result<Option<EncodedBulkPart>> {
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<RecordBatch>,
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());
}
+44 -1
View File
@@ -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,
})
+3
View File
@@ -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,
}
}
}
+3
View File
@@ -192,6 +192,7 @@ pub fn filter_record_batch(part: &BulkPart, min: i64, max: i64) -> Result<Option
max_timestamp: max_ts,
min_timestamp: min_ts,
sequence: part.sequence,
min_sequence: part.min_sequence,
timestamp_index: part.timestamp_index,
raw_data: None,
}))
@@ -1250,6 +1251,7 @@ mod tests {
max_timestamp: max_ts,
min_timestamp: min_ts,
sequence,
min_sequence: sequence,
timestamp_index: 0,
raw_data: None,
}
@@ -1388,6 +1390,7 @@ mod tests {
max_timestamp: 8000,
min_timestamp: 1000,
sequence: 0,
min_sequence: 0,
timestamp_index: 0,
raw_data: None,
};
+41 -1
View File
@@ -40,7 +40,7 @@ use mito_codec::key_values::KeyValue;
use mito_codec::row_converter::{DensePrimaryKeyCodec, PrimaryKeyCodecExt};
use snafu::{OptionExt, ResultExt, ensure};
use store_api::metadata::RegionMetadataRef;
use store_api::storage::{ColumnId, SequenceRange};
use store_api::storage::{ColumnId, SequenceNumber, SequenceRange};
use table::predicate::Predicate;
use crate::error::{
@@ -221,6 +221,7 @@ pub struct TimeSeriesMemtable {
max_timestamp: AtomicI64,
min_timestamp: AtomicI64,
max_sequence: AtomicU64,
min_sequence: AtomicU64,
dedup: bool,
merge_mode: MergeMode,
/// Total written rows in memtable. This also includes deleted and duplicated rows.
@@ -246,6 +247,7 @@ impl TimeSeriesMemtable {
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: Default::default(),
@@ -260,6 +262,8 @@ impl TimeSeriesMemtable {
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);
self.num_rows.fetch_add(stats.num_rows, Ordering::SeqCst);
}
@@ -291,6 +295,7 @@ impl TimeSeriesMemtable {
.value();
stats.min_ts = stats.min_ts.min(ts);
stats.max_ts = stats.max_ts.max(ts);
stats.min_sequence = stats.min_sequence.min(kv.sequence());
Ok(())
}
}
@@ -416,6 +421,7 @@ impl Memtable for TimeSeriesMemtable {
num_rows: 0,
num_ranges: 0,
max_sequence: 0,
min_sequence: 0,
series_count: 0,
};
}
@@ -436,10 +442,18 @@ impl Memtable for TimeSeriesMemtable {
num_rows: self.num_rows.load(Ordering::Relaxed),
num_ranges: 1,
max_sequence: self.max_sequence.load(Ordering::Relaxed),
min_sequence: self.min_sequence.load(Ordering::Relaxed),
series_count,
}
}
fn min_sequence(&self) -> SequenceNumber {
if self.alloc_tracker.bytes_allocated() == 0 {
return 0;
}
self.min_sequence.load(Ordering::Relaxed)
}
fn fork(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef {
Arc::new(TimeSeriesMemtable::new(
metadata.clone(),
@@ -1742,6 +1756,32 @@ mod tests {
KeyValues::new(schema.as_ref(), mutation).unwrap()
}
#[test]
fn test_min_sequence_covers_single_and_batch_writes_and_fork() {
let schema = schema_for_test();
let memtable =
TimeSeriesMemtable::new(schema.clone(), 1, None, true, MergeMode::LastNonNull);
let kvs = build_key_values(&schema, "a".to_string(), 1, 5);
assert_eq!(0, memtable.min_sequence());
for sequence in [4, 2] {
memtable
.write_one(kvs.iter().nth(sequence).unwrap())
.unwrap();
assert_eq!(sequence as u64, memtable.stats().min_sequence);
assert_eq!(sequence as u64, memtable.min_sequence());
}
memtable.write(&kvs).unwrap();
assert_eq!(0, memtable.stats().min_sequence);
assert_eq!(0, memtable.min_sequence());
assert_eq!(4, memtable.stats().max_sequence);
let fork = memtable.fork(2, &schema);
assert!(fork.is_empty());
assert_eq!(0, fork.min_sequence());
fork.write_one(kvs.iter().nth(3).unwrap()).unwrap();
assert_eq!(3, fork.stats().min_sequence);
assert_eq!(3, fork.min_sequence());
}
#[test]
fn test_series_set_concurrency() {
let schema = schema_for_test();
+11
View File
@@ -20,6 +20,7 @@ use std::time::Duration;
use smallvec::SmallVec;
use store_api::metadata::RegionMetadataRef;
use store_api::storage::SequenceNumber;
use crate::error::Result;
use crate::memtable::bulk::part::BulkPart;
@@ -74,6 +75,16 @@ impl MemtableVersion {
mems
}
/// Returns a sequence lower bound covering mutable and immutable memtables.
/// Empty memtables impose no compaction barrier, including newly forked ones.
pub(crate) fn min_sequence(&self) -> Option<SequenceNumber> {
self.list_memtables()
.iter()
.filter(|mem| !mem.is_empty())
.map(|mem| mem.min_sequence())
.min()
}
/// Returns a new [MemtableVersion] which switches the old mutable memtable to immutable
/// memtable.
///
+105 -58
View File
@@ -160,8 +160,8 @@ impl<I, S> Drop for FlatDedupReader<I, S> {
/// Strategy to remove duplicate rows from sorted record batches.
pub trait RecordBatchDedupStrategy: Send {
/// Pushes a batch to the dedup strategy.
/// Returns a batch if the strategy ensures there is no duplications based on
/// the input batch.
/// Returns rows that no longer need to be updated by subsequent input.
/// A strategy for partial merges may retain duplicates separated by Deletes.
fn push_batch(
&mut self,
batch: RecordBatch,
@@ -297,11 +297,11 @@ impl RecordBatchDedupStrategy for FlatLastRow {
pub struct FlatLastNonNull {
/// The start index of field columns:
field_column_start: usize,
/// Filter deleted rows.
/// Filter deleted rows; otherwise preserve every delete barrier.
filter_deleted: bool,
/// Buffered batch to check whether the next batch have duplicated rows with this batch.
/// Fields in the last row of this batch may be updated by the next batch.
/// The buffered batch should contain no duplication.
/// In preserve mode, duplicate keys may be separated by delete markers.
buffer: Option<BatchLastRow>,
/// Whether the last row range contains a delete operation.
/// If so, we don't need to update null fields.
@@ -326,7 +326,7 @@ impl RecordBatchDedupStrategy for FlatLastNonNull {
// If the buffer is None, dedup the batch, put the batch into the buffer and return.
// There is no previous batch with the same key, we can pass contains_delete as false.
let (record_batch, contains_delete) =
Self::dedup_one_batch(batch, self.field_column_start, false)?;
Self::dedup_one_batch(batch, self.field_column_start, false, self.filter_deleted)?;
metrics.num_unselected_rows += row_before_dedup - record_batch.num_rows();
self.buffer = BatchLastRow::try_new(record_batch);
self.contains_delete = contains_delete;
@@ -341,7 +341,7 @@ impl RecordBatchDedupStrategy for FlatLastNonNull {
// Dedup the batch.
// There is no previous batch with the same key, we can pass contains_delete as false.
let (record_batch, contains_delete) =
Self::dedup_one_batch(batch, self.field_column_start, false)?;
Self::dedup_one_batch(batch, self.field_column_start, false, self.filter_deleted)?;
metrics.num_unselected_rows += row_before_dedup - record_batch.num_rows();
debug_assert!(record_batch.num_rows() > 0);
self.buffer = BatchLastRow::try_new(record_batch);
@@ -369,8 +369,12 @@ impl RecordBatchDedupStrategy for FlatLastNonNull {
let merged = concat_batches(&schema, &[last_row, batch]).context(ComputeArrowSnafu)?;
let merged_row_count = merged.num_rows();
// Dedup the merged batch and update the buffer.
let (record_batch, contains_delete) =
Self::dedup_one_batch(merged, self.field_column_start, self.contains_delete)?;
let (record_batch, contains_delete) = Self::dedup_one_batch(
merged,
self.field_column_start,
self.contains_delete,
self.filter_deleted,
)?;
metrics.num_unselected_rows += merged_row_count - record_batch.num_rows();
debug_assert!(record_batch.num_rows() > 0);
self.buffer = BatchLastRow::try_new(record_batch);
@@ -397,7 +401,11 @@ impl RecordBatchDedupStrategy for FlatLastNonNull {
}
impl FlatLastNonNull {
/// Creates a new strategy with the given `filter_deleted` flag.
/// Creates a strategy with the given `filter_deleted` flag. When false,
/// every delete barrier is preserved and Puts are folded only between
/// barriers so later reads cannot fill fields from a deleted row.
/// Inputs for partial merges must form a sequence-contiguous group to avoid
/// folding fields across an intermediate version left outside the merge.
pub fn new(field_column_start: usize, filter_deleted: bool) -> Self {
Self {
field_column_start,
@@ -413,6 +421,7 @@ impl FlatLastNonNull {
batch: RecordBatch,
field_column_start: usize,
prev_batch_contains_delete: bool,
filter_deleted: bool,
) -> Result<(RecordBatch, bool)> {
// Get op type array for checking delete operations
let op_type_column = batch
@@ -460,6 +469,7 @@ impl FlatLastNonNull {
field_column_start,
op_types,
prev_batch_contains_delete,
filter_deleted,
)
}
@@ -471,9 +481,29 @@ impl FlatLastNonNull {
field_column_start: usize,
op_types: &UInt8Array,
first_range_contains_delete: bool,
filter_deleted: bool,
) -> Result<(RecordBatch, bool)> {
let ranges = partitions.ranges();
let contains_delete = Self::last_range_has_delete(&ranges, op_types);
let mut ranges = partitions.ranges();
let contains_delete = filter_deleted && Self::last_range_has_delete(&ranges, op_types);
if !filter_deleted && op_types.values().contains(&(OpType::Delete as u8)) {
let mut segments = Vec::with_capacity(ranges.len());
for range in ranges {
let mut start = range.start;
for i in range.clone() {
if op_types.value(i) == OpType::Delete as u8 {
if start < i {
segments.push(start..i);
}
segments.push(i..i + 1);
start = i + 1;
}
}
if start < range.end {
segments.push(start..range.end);
}
}
ranges = segments;
}
// Each range at least has 1 row.
let num_duplications: usize = ranges.iter().map(|r| r.end - r.start - 1).sum();
@@ -498,7 +528,7 @@ impl FlatLastNonNull {
&ranges,
column,
op_types,
first_range_contains_delete,
first_range_contains_delete && filter_deleted,
);
take(column, &field_indices, take_options.clone()).context(ComputeArrowSnafu)
} else {
@@ -1160,52 +1190,6 @@ mod tests {
check_record_batches_equal(&expected_filter_deleted, &result);
assert_eq!(6, dedup_iter.metrics.num_unselected_rows);
assert_eq!(2, dedup_iter.metrics.num_deleted_rows);
// Test with filter_deleted = false
let expected_no_filter = vec![
new_record_batch_multi_fields(
&[b"k1"],
&[1],
&[13],
&[OpType::Put],
&[(Some(11), Some(11))],
),
new_record_batch_multi_fields(
&[b"k1", b"k1", b"k1"],
&[2, 3, 4],
&[11, 13, 13],
&[OpType::Put, OpType::Put, OpType::Delete],
&[(Some(12), Some(22)), (Some(13), None), (None, Some(14))],
),
new_record_batch_multi_fields(
&[b"k2"],
&[1],
&[20],
&[OpType::Put],
&[(Some(101), Some(101))],
),
new_record_batch_multi_fields(
&[b"k2"],
&[2],
&[20],
&[OpType::Delete],
&[(None, None)],
),
new_record_batch_multi_fields(
&[b"k3"],
&[2],
&[20],
&[OpType::Put],
&[(Some(202), Some(202))],
),
];
let iter = input.clone().into_iter().map(Ok);
let mut dedup_iter = FlatDedupIterator::new(iter, FlatLastNonNull::new(1, false));
let result = collect_iterator_results(&mut dedup_iter);
check_record_batches_equal(&expected_no_filter, &result);
assert_eq!(4, dedup_iter.metrics.num_unselected_rows);
assert_eq!(0, dedup_iter.metrics.num_deleted_rows);
}
#[test]
@@ -1509,6 +1493,69 @@ mod tests {
);
}
#[test]
fn test_last_non_null_preserves_delete_barriers() {
let input = new_record_batch_multi_fields(
&[b"k1".as_slice(); 9],
&[1; 9],
&[9, 8, 7, 6, 5, 4, 3, 2, 1],
&[
OpType::Put,
OpType::Put,
OpType::Delete,
OpType::Delete,
OpType::Put,
OpType::Put,
OpType::Delete,
OpType::Put,
OpType::Put,
],
&[
(None, Some(9)),
(Some(8), None),
(None, None),
(None, None),
(None, Some(5)),
(Some(4), None),
(None, None),
(None, None),
(Some(1), Some(1)),
],
);
let expected = new_record_batch_multi_fields(
&[b"k1".as_slice(); 6],
&[1; 6],
&[9, 7, 6, 5, 3, 2],
&[
OpType::Put,
OpType::Delete,
OpType::Delete,
OpType::Put,
OpType::Delete,
OpType::Put,
],
&[
(Some(8), Some(9)),
(None, None),
(None, None),
(Some(4), Some(5)),
(None, None),
(Some(1), Some(1)),
],
);
// Check both intra-batch partitioning and barriers split across batches.
for batch_size in [1, input.num_rows()] {
let batches = (0..input.num_rows())
.step_by(batch_size)
.map(|start| Ok(input.slice(start, batch_size.min(input.num_rows() - start))));
let output = FlatDedupIterator::new(batches, FlatLastNonNull::new(1, false))
.collect::<Result<Vec<_>>>()
.unwrap();
let actual = concat_batches(&input.schema(), &output).unwrap();
assert_eq!(expected, actual);
}
}
#[test]
fn test_flat_last_non_null_strategy_delete_middle() {
let input = vec![
+3
View File
@@ -360,6 +360,7 @@ impl RegionWriteCtx {
self.next_sequence = sequence;
}
bulk.sequence = self.next_sequence;
bulk.min_sequence = self.next_sequence;
if !skip_wal {
let entry = match BulkWalEntry::try_from(&bulk) {
Ok(entry) => entry,
@@ -653,6 +654,7 @@ mod tests {
usize::from(!bulk_skip_wal)
);
assert_eq!(ctx.bulk_parts[0].sequence, 7);
assert_eq!(ctx.bulk_parts[0].min_sequence, 7);
let sequences: Vec<_> = ctx.wal_entry.mutations.iter().map(|m| m.sequence).collect();
assert_eq!(sequences, if skip_wal { vec![3, 6] } else { vec![1, 3, 6] });
// Check the actual WAL bytes after routing the mutations.
@@ -985,6 +987,7 @@ mod tests {
max_timestamp: 2,
min_timestamp: 1,
sequence: 0,
min_sequence: 0,
timestamp_index: 0,
raw_data: None,
}
+4
View File
@@ -99,6 +99,10 @@ impl Memtable for EmptyMemtable {
MemtableStats::default().with_time_range(self.time_range)
}
fn min_sequence(&self) -> SequenceNumber {
0
}
fn fork(&self, id: MemtableId, _metadata: &RegionMetadataRef) -> MemtableRef {
Arc::new(EmptyMemtable::new(id))
}
@@ -83,6 +83,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
max_timestamp: max_ts,
min_timestamp: min_ts,
sequence: 0,
min_sequence: 0,
timestamp_index: ts_index,
raw_data: Some(request.raw_data),
};
+1
View File
@@ -790,6 +790,7 @@ mod tests {
max_timestamp: num_rows - 1,
min_timestamp: 0,
sequence: 0,
min_sequence: 0,
timestamp_index: 1,
raw_data: None,
}