mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-21 11:22:31 +00:00
perf(mito2): keep short series unsplit in compaction merges (#9706)
* perf(mito2): keep short series unsplit in compaction merges Compaction writes every merged batch to the output SST. Splitting short series before the merge made it emit one small batch per series and input file, which the SST writer and indexer then handled one by one. Compaction now splits only when series are long enough to fill most of a batch. With 16 L0 files of 10k series x 30 rows, one output went from 160,560 merged batches to 592, and merging plus writing took 1.66s instead of 2.67s. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> * test(mito2): benchmark compacting L0 files into one output The benchmark writes 16 L0 files through the engine, each holding every series over its own time span as consecutive flushes do, and times a strict window compaction that merges them into one output. Every iteration compacts a fresh copy of the same files. With 10k series of 30 rows per file, keeping short series unsplit in compaction merges cut the time from 830ms to 520ms. With 60 series of 5,000 rows per file, which compaction still splits, both took 266ms. Signed-off-by: Lei, HUANG <ratuthomm@gmail.com> --------- Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
@@ -154,3 +154,8 @@ required-features = ["testing"]
|
||||
name = "bench_dense_index_update"
|
||||
harness = false
|
||||
required-features = ["testing"]
|
||||
|
||||
[[bench]]
|
||||
name = "bench_compaction_merge"
|
||||
harness = false
|
||||
required-features = ["test"]
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
// 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.
|
||||
|
||||
//! Benchmarks compacting L0 files into one output through the engine.
|
||||
//!
|
||||
//! Every input file holds all series over its own time span, as consecutive
|
||||
//! flushes do. Short series are where splitting the inputs by series hands the
|
||||
//! SST writer one small batch per series and file.
|
||||
//!
|
||||
//! Run with:
|
||||
//! ```sh
|
||||
//! cargo bench -p mito2 --features test --bench bench_compaction_merge
|
||||
//! ```
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use api::v1::Rows;
|
||||
use api::v1::region::{StrictWindow, compact_request};
|
||||
use common_test_util::temp_dir::create_temp_dir;
|
||||
use criterion::{Criterion, SamplingMode, Throughput, criterion_group, criterion_main};
|
||||
use mito2::config::MitoConfig;
|
||||
use mito2::engine::MitoEngine;
|
||||
use mito2::test_util::{
|
||||
CreateRequestBuilder, TestEnv, build_rows_for_key, flush_region, rows_schema,
|
||||
};
|
||||
use store_api::region_engine::{RegionEngine, RegionRole};
|
||||
use store_api::region_request::{
|
||||
PathType, RegionCompactRequest, RegionOpenRequest, RegionPutRequest, RegionRequest,
|
||||
};
|
||||
use store_api::storage::RegionId;
|
||||
|
||||
const REGION_ID: RegionId = RegionId::new(1, 1);
|
||||
/// Series written by each put request.
|
||||
const SERIES_PER_PUT: usize = 1000;
|
||||
/// The inputs of every shape fit in one day, so each flush writes one file and
|
||||
/// one compaction window covers all of them.
|
||||
const WINDOW_SECONDS: i64 = 24 * 3600;
|
||||
|
||||
struct Shape {
|
||||
name: &'static str,
|
||||
num_files: usize,
|
||||
num_series: usize,
|
||||
rows_per_series: usize,
|
||||
}
|
||||
|
||||
impl Shape {
|
||||
fn num_rows(&self) -> usize {
|
||||
self.num_files * self.num_series * self.rows_per_series
|
||||
}
|
||||
}
|
||||
|
||||
const SHAPES: [Shape; 2] = [
|
||||
// Few enough series for scans to split the inputs, each far shorter than a batch.
|
||||
Shape {
|
||||
name: "10k_series_x_30_rows",
|
||||
num_files: 16,
|
||||
num_series: 10_000,
|
||||
rows_per_series: 30,
|
||||
},
|
||||
// Series long enough to fill most of a batch.
|
||||
Shape {
|
||||
name: "60_series_x_5000_rows",
|
||||
num_files: 16,
|
||||
num_series: 60,
|
||||
rows_per_series: 5_000,
|
||||
},
|
||||
];
|
||||
|
||||
/// L0 files of a shape and the options to open the region holding them.
|
||||
struct Inputs {
|
||||
env: TestEnv,
|
||||
table_dir: String,
|
||||
path_type: PathType,
|
||||
options: HashMap<String, String>,
|
||||
}
|
||||
|
||||
impl Inputs {
|
||||
/// Opens a copy of the inputs in a new engine, so every compaction starts from the same files.
|
||||
async fn open_copy(&self) -> (TestEnv, MitoEngine) {
|
||||
let data_home = create_temp_dir("bench_compaction_merge");
|
||||
copy_dir(self.env.data_home(), data_home.path());
|
||||
let mut env = TestEnv::with_data_home(data_home).await;
|
||||
let engine = env.create_engine(engine_config()).await;
|
||||
let request = RegionOpenRequest {
|
||||
engine: String::new(),
|
||||
table_dir: self.table_dir.clone(),
|
||||
path_type: self.path_type,
|
||||
options: self.options.clone(),
|
||||
skip_wal_replay: false,
|
||||
checkpoint: None,
|
||||
requirements: Default::default(),
|
||||
};
|
||||
engine
|
||||
.handle_request(REGION_ID, RegionRequest::Open(request))
|
||||
.await
|
||||
.unwrap();
|
||||
engine
|
||||
.set_region_role(REGION_ID, RegionRole::Leader)
|
||||
.unwrap();
|
||||
(env, engine)
|
||||
}
|
||||
}
|
||||
|
||||
fn engine_config() -> MitoConfig {
|
||||
MitoConfig {
|
||||
// Keeps automatic compaction from merging the inputs.
|
||||
min_compaction_interval: Duration::from_secs(3600),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// Writes `shape.num_files` L0 files, each with every series over its own time span.
|
||||
async fn write_inputs(shape: &Shape) -> Inputs {
|
||||
let mut env = TestEnv::new().await;
|
||||
let engine = env.create_engine(engine_config()).await;
|
||||
let request = CreateRequestBuilder::new().build();
|
||||
let schema = rows_schema(&request);
|
||||
let table_dir = request.table_dir.clone();
|
||||
let path_type = request.path_type;
|
||||
let options = request.options.clone();
|
||||
engine
|
||||
.handle_request(REGION_ID, RegionRequest::Create(request))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
for file in 0..shape.num_files {
|
||||
let start = file * shape.rows_per_series;
|
||||
let end = start + shape.rows_per_series;
|
||||
for first in (0..shape.num_series).step_by(SERIES_PER_PUT) {
|
||||
let last = (first + SERIES_PER_PUT).min(shape.num_series);
|
||||
let rows = (first..last)
|
||||
.flat_map(|series| build_rows_for_key(&format!("host-{series:05}"), start, end, 0))
|
||||
.collect();
|
||||
let request = RegionPutRequest {
|
||||
rows: Rows {
|
||||
schema: schema.clone(),
|
||||
rows,
|
||||
},
|
||||
// The rows only need to reach the SSTs.
|
||||
skip_wal: true,
|
||||
hint: None,
|
||||
partition_expr_version: None,
|
||||
};
|
||||
engine
|
||||
.handle_request(REGION_ID, RegionRequest::Put(request))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
flush_region(&engine, REGION_ID, None).await;
|
||||
}
|
||||
let statistic = engine.get_region_statistic(REGION_ID).unwrap();
|
||||
assert_eq!(
|
||||
shape.num_files as u64, statistic.sst_num,
|
||||
"every flush must leave its own input file"
|
||||
);
|
||||
engine.stop().await.unwrap();
|
||||
|
||||
Inputs {
|
||||
env,
|
||||
table_dir,
|
||||
path_type,
|
||||
options,
|
||||
}
|
||||
}
|
||||
|
||||
fn copy_dir(from: &Path, to: &Path) {
|
||||
std::fs::create_dir_all(to).unwrap();
|
||||
for entry in std::fs::read_dir(from).unwrap() {
|
||||
let entry = entry.unwrap();
|
||||
let target = to.join(entry.file_name());
|
||||
if entry.file_type().unwrap().is_dir() {
|
||||
copy_dir(&entry.path(), &target);
|
||||
} else {
|
||||
std::fs::copy(entry.path(), target).unwrap();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn compact_request() -> RegionRequest {
|
||||
RegionRequest::Compact(RegionCompactRequest {
|
||||
options: compact_request::Options::StrictWindow(StrictWindow {
|
||||
window_seconds: WINDOW_SECONDS,
|
||||
}),
|
||||
..Default::default()
|
||||
})
|
||||
}
|
||||
|
||||
fn bench_compaction_merge(c: &mut Criterion) {
|
||||
let runtime = tokio::runtime::Runtime::new().unwrap();
|
||||
let mut group = c.benchmark_group("compaction_merge");
|
||||
group.sample_size(10);
|
||||
group.sampling_mode(SamplingMode::Flat);
|
||||
group.warm_up_time(Duration::from_secs(1));
|
||||
group.measurement_time(Duration::from_secs(20));
|
||||
for shape in &SHAPES {
|
||||
let inputs = runtime.block_on(write_inputs(shape));
|
||||
group.throughput(Throughput::Elements(shape.num_rows() as u64));
|
||||
group.bench_function(shape.name, |b| {
|
||||
b.iter_custom(|iterations| {
|
||||
runtime.block_on(async {
|
||||
let mut elapsed = Duration::ZERO;
|
||||
for _ in 0..iterations {
|
||||
let (_env, engine) = inputs.open_copy().await;
|
||||
let start = Instant::now();
|
||||
engine
|
||||
.handle_request(REGION_ID, compact_request())
|
||||
.await
|
||||
.unwrap();
|
||||
elapsed += start.elapsed();
|
||||
let statistic = engine.get_region_statistic(REGION_ID).unwrap();
|
||||
assert_eq!(1, statistic.sst_num, "inputs must merge into one output");
|
||||
engine.stop().await.unwrap();
|
||||
}
|
||||
elapsed
|
||||
})
|
||||
});
|
||||
});
|
||||
}
|
||||
group.finish();
|
||||
}
|
||||
|
||||
criterion_group!(benches, bench_compaction_merge);
|
||||
criterion_main!(benches);
|
||||
@@ -242,3 +242,81 @@ fn ts_to_lit(ts: Timestamp, ts_col_unit: TimeUnit) -> Result<Expr> {
|
||||
};
|
||||
Ok(datafusion_expr::lit(scalar_value))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use api::v1::Rows;
|
||||
use store_api::region_engine::RegionEngine;
|
||||
use store_api::region_request::{RegionFlushRequest, RegionRequest};
|
||||
use store_api::storage::RegionId;
|
||||
|
||||
use super::*;
|
||||
use crate::test_util::{
|
||||
CreateRequestBuilder, TestEnv, build_rows_for_key, put_rows, rows_schema,
|
||||
};
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_compaction_reader_keeps_short_series_batches_whole() {
|
||||
let mut env = TestEnv::new().await;
|
||||
let engine = env.create_engine(Default::default()).await;
|
||||
let region_id = RegionId::new(1, 1);
|
||||
let request = CreateRequestBuilder::new().build();
|
||||
let schema = rows_schema(&request);
|
||||
engine
|
||||
.handle_request(region_id, RegionRequest::Create(request))
|
||||
.await
|
||||
.unwrap();
|
||||
// Each file has 100 short series, which queries split for merging.
|
||||
for file in 0..2 {
|
||||
let rows = (0..100)
|
||||
.flat_map(|series| {
|
||||
build_rows_for_key(&format!("key-{series:03}"), file * 10, file * 10 + 10, 0)
|
||||
})
|
||||
.collect();
|
||||
put_rows(
|
||||
&engine,
|
||||
region_id,
|
||||
Rows {
|
||||
schema: schema.clone(),
|
||||
rows,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
engine
|
||||
.handle_request(
|
||||
region_id,
|
||||
RegionRequest::Flush(RegionFlushRequest::default()),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
let region = engine.get_region(region_id).unwrap();
|
||||
let version = region.version();
|
||||
let inputs: Vec<_> = version.ssts.levels()[0].files().cloned().collect();
|
||||
assert_eq!(2, inputs.len());
|
||||
|
||||
let mut source = CompactionSstReaderBuilder {
|
||||
metadata: version.metadata.clone(),
|
||||
sst_layer: region.access_layer.clone(),
|
||||
cache: engine.cache_manager(),
|
||||
inputs: &inputs,
|
||||
append_mode: false,
|
||||
filter_deleted: true,
|
||||
time_range: None,
|
||||
merge_mode: MergeMode::LastRow,
|
||||
}
|
||||
.build_flat_sst_reader()
|
||||
.await
|
||||
.unwrap();
|
||||
let mut num_batches = 0;
|
||||
let mut num_rows = 0;
|
||||
while let Some(batch) = source.next_batch().await.unwrap() {
|
||||
num_batches += 1;
|
||||
num_rows += batch.num_rows();
|
||||
}
|
||||
|
||||
// Splitting by series would yield a batch per series and file, 200 in total.
|
||||
assert_eq!(2000, num_rows);
|
||||
assert!(num_batches < 20, "got {num_batches} batches");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1204,6 +1204,9 @@ const NUM_SERIES_THRESHOLD: u64 = 10240;
|
||||
/// Minimum batch size after splitting. The batch size is less than 60 because a series may only have
|
||||
/// 60 samples per hour.
|
||||
const BATCH_SIZE_THRESHOLD: u64 = 50;
|
||||
/// Minimum estimated rows per series for compaction to split batches. Shorter series
|
||||
/// would make compaction write one small batch per series.
|
||||
pub(crate) const MIN_COMPACTION_SPLIT_ROWS: usize = DEFAULT_READ_BATCH_SIZE / 2;
|
||||
|
||||
/// Returns the estimated rows per batch after splitting if splitting flat record batches
|
||||
/// may improve merge performance. Returns `None` if splitting is not beneficial.
|
||||
|
||||
@@ -46,8 +46,9 @@ use crate::read::range_cache::{
|
||||
};
|
||||
use crate::read::scan_region::{ScanInput, StreamContext};
|
||||
use crate::read::scan_util::{
|
||||
PartitionMetrics, PartitionMetricsList, SplitRecordBatchStream, compute_parallel_channel_size,
|
||||
scan_flat_file_ranges, scan_flat_mem_ranges, should_split_flat_batches_for_merge,
|
||||
MIN_COMPACTION_SPLIT_ROWS, PartitionMetrics, PartitionMetricsList, SplitRecordBatchStream,
|
||||
compute_parallel_channel_size, scan_flat_file_ranges, scan_flat_mem_ranges,
|
||||
should_split_flat_batches_for_merge,
|
||||
};
|
||||
use crate::read::stream::{ConvertBatchStream, ScanBatch, ScanBatchStream};
|
||||
use crate::read::{BoxedRecordBatchStream, ScannerMetrics, scan_util};
|
||||
@@ -739,7 +740,11 @@ pub(crate) async fn build_flat_sources(
|
||||
let needs_merge =
|
||||
ordered_sources.iter().flatten().count() > 1 || (compaction && stream_ctx.ranges.len() > 1);
|
||||
let split_batch_size = if needs_merge {
|
||||
should_split_flat_batches_for_merge(stream_ctx, range_meta)
|
||||
should_split_flat_batches_for_merge(stream_ctx, range_meta).filter(|rows_per_series| {
|
||||
// Compaction writes every merged batch out, so splitting short series
|
||||
// would hand the SST writer and indexer one tiny batch per series.
|
||||
!compaction || *rows_per_series >= MIN_COMPACTION_SPLIT_ROWS
|
||||
})
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user