diff --git a/src/mito2/Cargo.toml b/src/mito2/Cargo.toml index 02c134e964c..ba6cbe5d8cb 100644 --- a/src/mito2/Cargo.toml +++ b/src/mito2/Cargo.toml @@ -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"] diff --git a/src/mito2/benches/bench_compaction_merge.rs b/src/mito2/benches/bench_compaction_merge.rs new file mode 100644 index 00000000000..a4b9b6bc6fe --- /dev/null +++ b/src/mito2/benches/bench_compaction_merge.rs @@ -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, +} + +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); diff --git a/src/mito2/src/compaction/reader.rs b/src/mito2/src/compaction/reader.rs index 7fde251cb80..159ebe2f22a 100644 --- a/src/mito2/src/compaction/reader.rs +++ b/src/mito2/src/compaction/reader.rs @@ -242,3 +242,81 @@ fn ts_to_lit(ts: Timestamp, ts_col_unit: TimeUnit) -> Result { }; 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"); + } +} diff --git a/src/mito2/src/read/scan_util.rs b/src/mito2/src/read/scan_util.rs index 7224143f2c4..53967949387 100644 --- a/src/mito2/src/read/scan_util.rs +++ b/src/mito2/src/read/scan_util.rs @@ -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. diff --git a/src/mito2/src/read/seq_scan.rs b/src/mito2/src/read/seq_scan.rs index bbf6ff1bf92..24c35481669 100644 --- a/src/mito2/src/read/seq_scan.rs +++ b/src/mito2/src/read/seq_scan.rs @@ -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 };