diff --git a/.github/runner-scale-sets/query-regression/README.md b/.github/runner-scale-sets/query-regression/README.md index f7a9bdc173..fa8d436d2c 100644 --- a/.github/runner-scale-sets/query-regression/README.md +++ b/.github/runner-scale-sets/query-regression/README.md @@ -108,7 +108,7 @@ agent, which Aliyun public Ubuntu images include. An allowlisted repository admin commenting `/query-regression` on the PR is **trust admission for that exact PR revision**. `/query-regression` runs the -six routine default cases; `/query-regression heavy` runs only the +nine routine default cases; `/query-regression heavy` runs only the high-cardinality `prom_remote_write_7913` remote-write case. `slash-command-dispatch.yml` (`issue_comment` on the default branch) parses the command and `repository_dispatch`es; `query-regression-slash.yml` admits diff --git a/.github/scripts/query-regression-run.py b/.github/scripts/query-regression-run.py index a99706c04a..6da7c896d6 100644 --- a/.github/scripts/query-regression-run.py +++ b/.github/scripts/query-regression-run.py @@ -33,6 +33,7 @@ from typing import Any DEFAULT_CASES = [ "tests/perf/query_cases/smoke_direct_sst/case.toml", + "tests/perf/query_cases/sst_float_bss/case.toml", "tests/perf/query_cases/prom_remote_write_seeded_random/case.toml", "tests/perf/query_cases/prom_remote_write_run_heavy/case.toml", "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml", diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index a76c49d43d..0f5239a6b5 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -105,6 +105,10 @@ pub(super) struct PromRemoteWritePlan { #[serde(default = "default_visibility_timeout_seconds")] pub(super) visibility_timeout_seconds: u64, #[serde(default)] + pub(super) base_setup_sql: Vec, + #[serde(default)] + pub(super) candidate_setup_sql: Vec, + #[serde(default)] pub(super) prom_store: PromStoreConfig, #[serde(default)] pub(super) value: ValueConfig, @@ -200,6 +204,7 @@ pub(super) enum ValuePattern { QuantizedSignal, SignalWithSporadicStalls, MixedSignalRepeated, + BoundedMixed, } impl std::fmt::Display for ValuePattern { @@ -436,6 +441,63 @@ pub(super) fn default_parallelism() -> u64 { 1 } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn remote_write_setup_sql_defaults_to_empty() { + let case: CaseFile = toml::from_str( + r#" +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +metric = "metric" +"#, + ) + .unwrap(); + let Scenario::PromRemoteWriteThenQuery(scenario) = case.scenario else { + panic!("expected prom_remote_write_then_query scenario"); + }; + assert!(scenario.remote_write.base_setup_sql.is_empty()); + assert!(scenario.remote_write.candidate_setup_sql.is_empty()); + } + + #[test] + fn remote_write_setup_sql_roundtrips() { + let case: CaseFile = toml::from_str( + r#" +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +metric = "metric" +base_setup_sql = ["CREATE TABLE base_table"] +candidate_setup_sql = ["CREATE TABLE candidate_table", "ALTER TABLE candidate_table SET 'x'='y'"] +"#, + ) + .unwrap(); + + let roundtrip: CaseFile = + serde_json::from_str(&serde_json::to_string(&case).unwrap()).unwrap(); + let Scenario::PromRemoteWriteThenQuery(scenario) = roundtrip.scenario else { + panic!("expected prom_remote_write_then_query scenario"); + }; + assert_eq!( + scenario.remote_write.base_setup_sql, + ["CREATE TABLE base_table"] + ); + assert_eq!( + scenario.remote_write.candidate_setup_sql, + [ + "CREATE TABLE candidate_table", + "ALTER TABLE candidate_table SET 'x'='y'" + ] + ); + } +} + impl Scenario { pub(super) fn kind(&self) -> &'static str { match self { diff --git a/src/cmd/src/bin/query_perf_fixture/direct_sst.rs b/src/cmd/src/bin/query_perf_fixture/direct_sst.rs index 746ecbd76e..27a6b4109c 100644 --- a/src/cmd/src/bin/query_perf_fixture/direct_sst.rs +++ b/src/cmd/src/bin/query_perf_fixture/direct_sst.rs @@ -469,6 +469,7 @@ pub(super) async fn run_direct_sst(args: DirectArgs) { write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE, row_group_size: scenario.layout.row_group_size, max_file_size: None, + float_field_encoding: Default::default(), }; let infos = match format { FormatType::Flat => writer.write_all_flat(source, Some(sequence), &opts).await, diff --git a/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs b/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs index 1f36b0c2d2..a59178f4a7 100644 --- a/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs +++ b/src/cmd/src/bin/query_perf_fixture/prom_remote_write.rs @@ -197,8 +197,35 @@ fn deterministic_prom_value(args: &PromRemoteWriteArgs, series_idx: u64, sample_ args.value_base + (series_idx % 97) as f64 + local as f64 * args.value_step } } + ValuePattern::BoundedMixed => bounded_mixed_value(args, series_idx, local), } } + +fn bounded_mixed_value(args: &PromRemoteWriteArgs, series_idx: u64, local: u64) -> f64 { + const SPANS: [i64; 4] = [10, 1_000, 100_000, 10_000_000]; + let series = splitmix64(series_idx ^ args.value_seed); + let span = SPANS[(series >> 62) as usize] * (1 + ((series >> 32) % 10) as i64); + let baseline = (series % args.value_cardinality.max(1)) as f64 * span as f64 / 10.0; + let run_length = args.value_run_length.max(1); + let group = local / run_length; + let phase = local % run_length; + let anchor = bounded_mixed_anchor(series, group, span); + let next_anchor = bounded_mixed_anchor(series, group + 1, span); + let signal = baseline + anchor + (next_anchor - anchor) * phase as f64 / run_length as f64; + let value = (args.value_base + args.value_step * signal).round(); + if series_idx % args.value_mixed_every.max(1) == args.value_mixed_every.max(1) - 1 { + let numerator = splitmix64(series ^ local) % 999 + 1; + value + numerator as f64 / 1_000.0 + } else { + value + } +} + +fn bounded_mixed_anchor(series: u64, group: u64, span: i64) -> f64 { + let limit = span - 1; + (splitmix64(series ^ group) % (2 * limit as u64 + 1)) as i64 as f64 - limit as f64 +} + fn splitmix64(mut value: u64) -> u64 { value = value.wrapping_add(0x9e3779b97f4a7c15); let mut mixed = value; @@ -307,7 +334,7 @@ mod tests { for chunk_idx in 0..10 { let mut chunk = args(ValuePattern::SeededRandom); chunk.samples_per_series = 1_440; - chunk.value_sample_offset = chunk_idx * 1_440; + chunk.value_sample_offset = chunk_idx * 1_439; chunk.value_total_samples_per_series = Some(14_400); for sample_idx in 0..chunk.samples_per_series { assert_eq!( @@ -319,6 +346,118 @@ mod tests { } } + #[test] + fn bounded_mixed_has_bounded_95_5_per_series_values() { + let mut args = args(ValuePattern::BoundedMixed); + args.series_count = 1_000; + args.samples_per_series = 4_320; + args.value_cardinality = 1_000; + args.value_seed = 42; + args.value_run_length = 60; + args.value_mixed_every = 20; + + let mut integral_series = 0; + let mut fractional_series = 0; + let mut spans = HashSet::new(); + let mut ranges = HashSet::new(); + for series_idx in 0..args.series_count { + let series = splitmix64(series_idx ^ args.value_seed); + let span = [10_i64, 1_000, 100_000, 10_000_000][(series >> 62) as usize] + * (1 + ((series >> 32) % 10) as i64); + let baseline = (series % args.value_cardinality.max(1)) as f64 * span as f64 / 10.0; + let values = (0..args.samples_per_series) + .map(|sample_idx| deterministic_prom_value(&args, series_idx, sample_idx)) + .collect::>(); + + assert!(values.iter().all(|value| value.is_finite())); + assert!( + values + .iter() + .all(|value| *value >= baseline - span as f64 - 1.0 + && *value <= baseline + span as f64 + 1.0) + ); + assert!( + values.windows(2).any(|pair| pair[1] > pair[0]), + "series {series_idx} never rises" + ); + assert!( + values.windows(2).any(|pair| pair[1] < pair[0]), + "series {series_idx} never falls" + ); + ranges.insert(( + values + .iter() + .copied() + .fold(f64::INFINITY, f64::min) + .to_bits(), + values + .iter() + .copied() + .fold(f64::NEG_INFINITY, f64::max) + .to_bits(), + )); + spans.insert(span); + if series_idx % args.value_mixed_every == args.value_mixed_every - 1 { + fractional_series += 1; + assert!(values.iter().all(|value| value.fract() != 0.0)); + } else { + integral_series += 1; + assert!(values.iter().all(|value| value.fract() == 0.0)); + } + } + + assert_eq!((integral_series, fractional_series), (950, 50)); + assert!(spans.len() > 4); + assert!(ranges.len() > 1); + } + + #[test] + fn bounded_mixed_is_seeded_and_chunk_bit_identical() { + let mut monolithic = args(ValuePattern::BoundedMixed); + monolithic.samples_per_series = 14_400; + monolithic.value_total_samples_per_series = Some(14_400); + monolithic.value_cardinality = 1_000; + monolithic.value_seed = 42; + monolithic.value_run_length = 60; + monolithic.value_mixed_every = 20; + + let original = (0..600) + .map(|sample_idx| deterministic_prom_value(&monolithic, 19, sample_idx)) + .collect::>(); + monolithic.value_seed = 43; + assert_ne!( + original, + (0..600) + .map(|sample_idx| deterministic_prom_value(&monolithic, 19, sample_idx)) + .collect::>() + ); + + monolithic.value_seed = 42; + for series_idx in [0, 19] { + for chunk_idx in 0..10 { + let mut chunk = args(ValuePattern::BoundedMixed); + chunk.samples_per_series = 1_440; + chunk.value_sample_offset = chunk_idx * 1_439; + chunk.value_total_samples_per_series = Some(14_400); + chunk.value_cardinality = 1_000; + chunk.value_seed = 42; + chunk.value_run_length = 60; + chunk.value_mixed_every = 20; + for sample_idx in 0..chunk.samples_per_series { + assert_eq!( + deterministic_prom_value(&chunk, series_idx, sample_idx).to_bits(), + deterministic_prom_value( + &monolithic, + series_idx, + chunk.value_sample_offset + sample_idx + ) + .to_bits() + ); + } + } + } + } + #[test] fn values_continue_across_series_boundaries() { let mut args = args(ValuePattern::Unique); diff --git a/src/cmd/src/bin/query_regression_runner/model.rs b/src/cmd/src/bin/query_regression_runner/model.rs index 6831a5256a..5fafdb5635 100644 --- a/src/cmd/src/bin/query_regression_runner/model.rs +++ b/src/cmd/src/bin/query_regression_runner/model.rs @@ -116,6 +116,10 @@ pub(super) struct RemoteWrite { pub(super) sample_chunk_size: Option, pub(super) flush_every_sample_chunks: u64, pub(super) visibility_timeout_seconds: u64, + #[serde(default)] + pub(super) base_setup_sql: Vec, + #[serde(default)] + pub(super) candidate_setup_sql: Vec, pub(super) prom_store: PromStore, pub(super) value: RemoteValue, pub(super) storage: Option, diff --git a/src/cmd/src/bin/query_regression_runner/remote/ingest.rs b/src/cmd/src/bin/query_regression_runner/remote/ingest.rs index f563402afc..271e27fac4 100644 --- a/src/cmd/src/bin/query_regression_runner/remote/ingest.rs +++ b/src/cmd/src/bin/query_regression_runner/remote/ingest.rs @@ -57,6 +57,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> { args.base_http_port, &args.fixture_generator, &remote, + &remote.base_setup_sql, &client, ) .await?; @@ -65,6 +66,7 @@ pub(super) async fn run_prepare_remote(args: PrepareRemoteArgs) -> Result<()> { args.candidate_http_port, &args.fixture_generator, &remote, + &remote.candidate_setup_sql, &client, ) .await?; @@ -88,6 +90,7 @@ async fn prepare_remote_target( port: u16, generator: &Path, remote: &RemoteWrite, + setup_sql: &[String], client: &Client, ) -> Result { let create_database = http_post_sql( @@ -107,6 +110,7 @@ async fn prepare_remote_target( ) .into()); } + let setup = run_setup_sql(client, port, &remote.database, name, setup_sql).await?; let (remote_write, flushes) = ingest_remote_write(generator, port, remote, client).await?; let expected_rows = remote .series_count @@ -124,6 +128,7 @@ async fn prepare_remote_target( Ok(json!({ "name": name, "create_database": create_database, + "setup_sql": setup, "remote_write": remote_write, "flushes": flushes, "visibility": visibility, @@ -131,6 +136,28 @@ async fn prepare_remote_target( })) } +async fn run_setup_sql( + client: &Client, + port: u16, + database: &str, + target: &str, + setup_sql: &[String], +) -> Result> { + let mut results = Vec::with_capacity(setup_sql.len()); + for (index, statement) in setup_sql.iter().enumerate() { + let result = http_post_sql(client, port, statement, database).await; + if !result["ok"].as_bool().unwrap_or(false) { + return Err(format!( + "setup SQL statement {} failed for {target}: {result}", + index + 1 + ) + .into()); + } + results.push(result); + } + Ok(results) +} + #[derive(Clone, Debug, PartialEq, Eq)] struct SampleChunk { index: u64, @@ -434,9 +461,110 @@ fn json_number(value: f64) -> String { #[cfg(test)] mod tests { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; + use super::*; use crate::query_regression_runner::model::{PromStore, RemoteValue}; + async fn read_http_request(stream: &mut TcpStream) -> Vec { + let mut request = Vec::new(); + let mut chunk = [0; 1024]; + loop { + let read = stream.read(&mut chunk).await.unwrap(); + assert_ne!( + read, 0, + "HTTP client closed request before sending its body" + ); + request.extend_from_slice(&chunk[..read]); + let Some(header_end) = request.windows(4).position(|window| window == b"\r\n\r\n") + else { + continue; + }; + let headers = std::str::from_utf8(&request[..header_end]).unwrap(); + let content_length = headers + .lines() + .find_map(|line| line.strip_prefix("content-length: ")) + .or_else(|| { + headers + .lines() + .find_map(|line| line.strip_prefix("Content-Length: ")) + }) + .unwrap() + .parse::() + .unwrap(); + if request.len() >= header_end + 4 + content_length { + return request; + } + } + } + + async fn spawn_sql_server( + responses: Vec, + ) -> (u16, tokio::task::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let server = tokio::spawn(async move { + let mut requests = Vec::with_capacity(responses.len()); + for body in responses { + let (mut stream, _) = listener.accept().await.unwrap(); + let request = read_http_request(&mut stream).await; + requests.push(String::from_utf8(request).unwrap()); + let response = format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Connection: close\r\nContent-Length: {}\r\n\r\n{body}", + body.len() + ); + stream.write_all(response.as_bytes()).await.unwrap(); + } + requests + }); + (port, server) + } + + #[tokio::test] + async fn setup_sql_runs_in_order_with_configured_database() { + let (port, server) = spawn_sql_server(vec!["{\"code\":0}".to_string(); 2]).await; + let client = Client::new(); + let setup = vec![ + "CREATE TABLE physical (ts TIMESTAMP TIME INDEX) ENGINE=metric".to_string(), + "ALTER TABLE physical SET 'compaction.twcs.trigger_file_num'='100'".to_string(), + ]; + + let results = run_setup_sql(&client, port, "perf", "candidate", &setup) + .await + .unwrap(); + + assert_eq!(results.len(), 2); + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 2); + assert!(requests[0].contains("db=perf")); + assert!(requests[0].contains("sql=CREATE")); + assert!(requests[1].contains("db=perf")); + assert!(requests[1].contains("sql=ALTER")); + } + + #[tokio::test] + async fn setup_sql_stops_at_the_failing_statement() { + let (port, server) = + spawn_sql_server(vec!["{\"code\":0}".to_string(), "{\"code\":1}".to_string()]).await; + let client = Client::new(); + let setup = vec![ + "CREATE TABLE physical (ts TIMESTAMP TIME INDEX) ENGINE=metric".to_string(), + "invalid SQL".to_string(), + "ALTER TABLE physical SET 'compaction.twcs.trigger_file_num'='100'".to_string(), + ]; + + let error = run_setup_sql(&client, port, "perf", "base", &setup) + .await + .unwrap_err(); + + assert!(error.to_string().contains("statement 2 failed for base")); + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 2); + assert!(requests[1].contains("sql=invalid+SQL")); + } + #[test] fn schedules_remote_sample_chunks_and_flushes() { let remote = RemoteWrite { @@ -452,6 +580,8 @@ mod tests { sample_chunk_size: Some(2), flush_every_sample_chunks: 2, visibility_timeout_seconds: 30, + base_setup_sql: Vec::new(), + candidate_setup_sql: Vec::new(), prom_store: PromStore { pending_rows_flush_interval: "1s".to_string(), max_batch_rows: 1, diff --git a/src/cmd/src/bin/query_regression_runner/remote/read_bench.rs b/src/cmd/src/bin/query_regression_runner/remote/read_bench.rs index 6b20a31c18..b9a05f6a69 100644 --- a/src/cmd/src/bin/query_regression_runner/remote/read_bench.rs +++ b/src/cmd/src/bin/query_regression_runner/remote/read_bench.rs @@ -306,13 +306,25 @@ fn run_bench_command(command: Vec, mut run: Value) -> Result { } fn parse_average_duration(stdout: &str) -> Option { - Regex::new(r"(?i)Average duration[^0-9]*([0-9.]+)\s*ms") - .ok()? - .captures(stdout)? - .get(1)? - .as_str() - .parse() - .ok() + let captures = Regex::new( + r"(?i)Average:\s+(?:\x1b\[[0-9;]*m)*\d+(?:\x1b\[[0-9;]*m)*\s+rows(?:,\s+(?:\x1b\[[0-9;]*m)*\d+(?:\x1b\[[0-9;]*m)*\s+record batches)?\s+in\s+([0-9]+(?:\.[0-9]+)?)\s*(ns|µs|us|ms|s)\s+over\s+\d+\s+iterations|Average duration[^0-9]*([0-9]+(?:\.[0-9]+)?)\s*ms", + ) + .ok()? + .captures(stdout)?; + let (value, unit) = match (captures.get(1), captures.get(2), captures.get(3)) { + (Some(value), Some(unit), _) => (value.as_str(), unit.as_str()), + (_, _, Some(value)) => (value.as_str(), "ms"), + _ => return None, + }; + let value = value.parse::().ok()?; + let milliseconds = match unit.to_ascii_lowercase().as_str() { + "ns" => value / 1_000_000.0, + "µs" | "us" => value / 1_000.0, + "ms" => value, + "s" => value * 1_000.0, + _ => return None, + }; + milliseconds.is_finite().then_some(milliseconds) } fn bench_median(runs: &[Value]) -> Option { @@ -328,12 +340,51 @@ mod tests { use super::*; #[test] - fn parses_average_duration_and_groups_scan_paths() { - assert_eq!( - parse_average_duration("work\nAverage duration: 12.5 ms\n"), - Some(12.5) - ); - assert_eq!(parse_average_duration("completed"), None); + fn parses_average_durations() { + let cases = [ + ( + "parquetbench nanoseconds", + "ℹ Average: 42 rows, 2 record batches in 500ns over 4 iterations", + Some(0.0005), + ), + ( + "scanbench colored microseconds", + "\x1b[1;32mSummary\x1b[0m Average: \x1b[36m12\x1b[0m rows in 250µs over 2 iterations", + Some(0.25), + ), + ( + "scanbench seconds", + "Summary Average: 12 rows in 2.5s over 2 iterations", + Some(2500.0), + ), + ( + "legacy milliseconds", + "Average duration: 12.5 ms", + Some(12.5), + ), + ("absent", "Benchmark completed!", None), + ( + "malformed", + "Summary Average: 12 rows in unknown over 2 iterations", + None, + ), + ]; + + for (name, stdout, expected) in cases { + let actual = parse_average_duration(stdout); + match (actual, expected) { + (Some(actual), Some(expected)) => assert!( + (actual - expected).abs() < f64::EPSILON, + "{name}: expected {expected}, got {actual}" + ), + (None, None) => {} + (actual, expected) => panic!("{name}: expected {expected:?}, got {actual:?}"), + } + } + } + + #[test] + fn groups_scan_paths() { let targets = vec![ BenchTarget { relative_path: "data/a.parquet".to_string(), diff --git a/src/metric-engine/src/engine/bulk_insert.rs b/src/metric-engine/src/engine/bulk_insert.rs index af2122dabb..cb43a71406 100644 --- a/src/metric-engine/src/engine/bulk_insert.rs +++ b/src/metric-engine/src/engine/bulk_insert.rs @@ -242,18 +242,22 @@ mod tests { use common_error::ext::ErrorExt; use common_query::prelude::{greptime_timestamp, greptime_value}; use common_recordbatch::RecordBatches; + use datafusion::parquet::basic::Encoding; use datatypes::arrow::array::{Float64Array, StringArray, TimestampMillisecondArray}; use datatypes::arrow::datatypes::{DataType, Field, Schema as ArrowSchema, TimeUnit}; use datatypes::arrow::record_batch::RecordBatch; use mito2::config::MitoConfig; + use mito2::sst::parquet::metadata::MetadataLoader; + use mito2::sst::parquet::reader::MetadataCacheMetrics; use store_api::metric_engine_consts::{ METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY, PRIMARY_KEY_ENCODING, }; + use store_api::mito_engine_options::EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING; use store_api::path_utils::table_dir; use store_api::region_engine::RegionEngine; use store_api::region_request::{ - PathType, RegionBulkInsertsRequest, RegionCloseRequest, RegionOpenRequest, - RegionPutRequest, RegionRequest, + PathType, RegionBulkInsertsRequest, RegionCloseRequest, RegionFlushRequest, + RegionOpenRequest, RegionPutRequest, RegionRequest, }; use store_api::storage::{RegionId, ScanRequest}; @@ -611,16 +615,68 @@ mod tests { #[tokio::test] async fn test_bulk_insert_sparse_encoding() { let env = TestEnv::new().await; - env.init_metric_region().await; + let physical_region_id = env.default_physical_region_id(); + env.create_physical_region( + physical_region_id, + &TestEnv::default_table_dir(), + vec![( + EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING.to_string(), + "byte_stream_split".to_string(), + )], + ) + .await; let logical_region_id = env.default_logical_region_id(); + env.create_logical_region(physical_region_id, logical_region_id) + .await; - let request = build_bulk_request(logical_region_id, build_logical_batch(0, 4), false); + let rows = 4; + let request = build_bulk_request(logical_region_id, build_logical_batch(0, rows), false); let response = env .metric() .handle_request(logical_region_id, request) .await .unwrap(); - assert_eq!(response.affected_rows, 4); + assert_eq!(response.affected_rows, rows); + + let data_region_id = crate::utils::to_data_region_id(physical_region_id); + env.mito() + .handle_request( + data_region_id, + RegionRequest::Flush(RegionFlushRequest::default()), + ) + .await + .unwrap(); + + let region = env.mito().find_region(data_region_id).unwrap(); + let entry = region + .manifest_sst_entries() + .await + .into_iter() + .find(|entry| entry.visible && entry.file_path.ends_with(".parquet")) + .unwrap(); + let mut cache_metrics = MetadataCacheMetrics::default(); + let footer = MetadataLoader::new( + region.access_layer().object_store().clone(), + &entry.file_path, + entry.file_size, + ) + .load(&mut cache_metrics) + .await + .unwrap(); + let field_column_index = footer + .file_metadata() + .schema_descr() + .columns() + .iter() + .position(|column| column.name() == greptime_value()) + .unwrap(); + assert!(!footer.row_groups().is_empty()); + assert!(footer.row_groups().iter().all(|row_group| { + row_group + .column(field_column_index) + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + })); let stream = env .metric() @@ -628,7 +684,7 @@ mod tests { .await .unwrap(); let batches = RecordBatches::try_collect(stream).await.unwrap(); - assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), 4); + assert_eq!(batches.iter().map(|b| b.num_rows()).sum::(), rows); } #[tokio::test] diff --git a/src/mito2/src/compaction/compactor.rs b/src/mito2/src/compaction/compactor.rs index 8ad1cdf0b2..fb2172b487 100644 --- a/src/mito2/src/compaction/compactor.rs +++ b/src/mito2/src/compaction/compactor.rs @@ -655,6 +655,7 @@ where write_buffer_size: compaction_region.engine_config.sst_write_buffer_size, max_file_size: picker_output.max_file_size, row_group_size: compaction_region.region_options.row_group_size(), + float_field_encoding: compaction_region.region_options.float_field_encoding, }; let merger = self.merger.clone(); let compaction_region = compaction_region.clone(); diff --git a/src/mito2/src/compaction/window.rs b/src/mito2/src/compaction/window.rs index a9f3f021b8..b905346644 100644 --- a/src/mito2/src/compaction/window.rs +++ b/src/mito2/src/compaction/window.rs @@ -337,6 +337,7 @@ mod tests { primary_key_encoding: None, write_buffer_size: None, preserve_row_sequence: false, + float_field_encoding: Default::default(), }, compaction_time_window: None, } diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs index 79c88f8a6c..a9ee405415 100644 --- a/src/mito2/src/engine/basic_test.rs +++ b/src/mito2/src/engine/basic_test.rs @@ -29,6 +29,7 @@ use datatypes::prelude::ConcreteDataType; use datatypes::schema::ColumnSchema; use futures::TryStreamExt; use itertools::Itertools; +use parquet::basic::{Encoding, Type as PhysicalType}; use rstest::rstest; use rstest_reuse::{self, apply}; use store_api::metadata::ColumnMetadata; @@ -826,7 +827,9 @@ async fn test_engine_with_write_cache_with_format(flat_format: bool) { let engine = env.create_engine(mito_config).await; let region_id = RegionId::new(1, 1); - let request = CreateRequestBuilder::new().build(); + let request = CreateRequestBuilder::new() + .insert_option("experimental_sst_float_field_encoding", "byte_stream_split") + .build(); let column_schemas = rows_schema(&request); engine @@ -842,6 +845,38 @@ async fn test_engine_with_write_cache_with_format(flat_format: bool) { flush_region(&engine, region_id, None).await; + let region = engine.get_region(region_id).unwrap(); + let file = region + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .next() + .expect("write-cache flushed SST") + .clone(); + let reader = region + .access_layer + .read_sst(file) + .build() + .await + .unwrap() + .expect("write-cache flushed SST reader"); + assert!( + reader + .parquet_metadata() + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .any(|column| { + column.column_path().string() == "field_0" + && column.column_type() == PhysicalType::DOUBLE + && column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + }) + ); + let request = ScanRequest::default(); let scanner = engine.scanner(region_id, request).await.unwrap(); diff --git a/src/mito2/src/engine/compaction_test.rs b/src/mito2/src/engine/compaction_test.rs index c67afbf8c6..e427f3f34f 100644 --- a/src/mito2/src/engine/compaction_test.rs +++ b/src/mito2/src/engine/compaction_test.rs @@ -26,6 +26,7 @@ use common_recordbatch::{RecordBatches, SendableRecordBatchStream}; use common_time::Timestamp; use datatypes::arrow::array::AsArray; use datatypes::arrow::datatypes::TimestampMillisecondType; +use parquet::basic::{Encoding, Type as PhysicalType}; use store_api::region_engine::{RegionEngine, RegionRole}; use store_api::region_request::AlterKind::SetRegionOptions; use store_api::region_request::{ @@ -954,6 +955,7 @@ async fn test_compaction_region_with_format(flat_format: bool) { let request = CreateRequestBuilder::new() .insert_option("compaction.type", "twcs") + .insert_option("experimental_sst_float_field_encoding", "byte_stream_split") .build(); let column_schemas = request @@ -974,6 +976,35 @@ async fn test_compaction_region_with_format(flat_format: bool) { compact(&engine, region_id).await; + let region = engine.get_region(region_id).unwrap(); + let file = region.version().ssts.levels()[1] + .files + .values() + .next() + .expect("compaction output SST") + .clone(); + let reader = region + .access_layer + .read_sst(file) + .build() + .await + .unwrap() + .expect("compaction output SST reader"); + assert!( + reader + .parquet_metadata() + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .any(|column| { + column.column_path().string() == "field_0" + && column.column_type() == PhysicalType::DOUBLE + && column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + }) + ); + let scanner = engine .scanner(region_id, ScanRequest::default()) .await diff --git a/src/mito2/src/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index d7155e1052..2b1e6f0cc2 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -27,6 +27,7 @@ use common_base::readable_size::ReadableSize; use common_recordbatch::RecordBatches; use common_time::util::current_time_millis; use common_wal::options::{KafkaWalOptions, WAL_OPTIONS_KEY, WalOptions}; +use parquet::basic::{Encoding, Type as PhysicalType}; use rstest::rstest; use rstest_reuse::{self, apply}; use store_api::ManifestVersion; @@ -279,7 +280,9 @@ async fn test_manual_flush_with_format(flat_format: bool) { ) .await; - let request = CreateRequestBuilder::new().build(); + let request = CreateRequestBuilder::new() + .insert_option("experimental_sst_float_field_encoding", "byte_stream_split") + .build(); let column_schemas = rows_schema(&request); engine @@ -295,6 +298,38 @@ async fn test_manual_flush_with_format(flat_format: bool) { flush_region(&engine, region_id, None).await; + let region = engine.get_region(region_id).unwrap(); + let file = region + .version() + .ssts + .levels() + .iter() + .flat_map(|level| level.files.values()) + .next() + .expect("flushed SST") + .clone(); + let reader = region + .access_layer + .read_sst(file) + .build() + .await + .unwrap() + .expect("flushed SST reader"); + assert!( + reader + .parquet_metadata() + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .any(|column| { + column.column_path().string() == "field_0" + && column.column_type() == PhysicalType::DOUBLE + && column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + }) + ); + let request = ScanRequest::default(); let scanner = engine.scanner(region_id, request).await.unwrap(); assert_eq!(0, scanner.num_memtables()); diff --git a/src/mito2/src/flush.rs b/src/mito2/src/flush.rs index cf7c6c0285..7b431c1b89 100644 --- a/src/mito2/src/flush.rs +++ b/src/mito2/src/flush.rs @@ -451,6 +451,7 @@ impl RegionFlushTask { let mut write_opts = WriteOptions { write_buffer_size: self.engine_config.sst_write_buffer_size, + float_field_encoding: version.options.float_field_encoding, ..Default::default() }; if let Some(row_group_size) = self.row_group_size { diff --git a/src/mito2/src/memtable.rs b/src/mito2/src/memtable.rs index f6481fbcdc..bc085c3bd4 100644 --- a/src/mito2/src/memtable.rs +++ b/src/mito2/src/memtable.rs @@ -491,6 +491,7 @@ impl MemtableBuilderProvider { BulkMemtableBuilder::new(self.write_buffer_manager.clone(), !dedup, merge_mode) .with_config(config.clone()) .with_row_group_size(options.row_group_size()) + .with_float_field_encoding(options.float_field_encoding) .with_compact_dispatcher(self.compact_dispatcher.clone()), ), Some(MemtableOptions::TimeSeries) => Arc::new(TimeSeriesMemtableBuilder::new( @@ -515,6 +516,7 @@ impl MemtableBuilderProvider { ) .with_config(self.default_bulk_memtable_config.clone()) .with_row_group_size(options.row_group_size()) + .with_float_field_encoding(options.float_field_encoding) .with_compact_dispatcher(self.compact_dispatcher.clone()); if let Some(MemtableOptions::Bulk(config)) = &options.memtable { diff --git a/src/mito2/src/memtable/bulk.rs b/src/mito2/src/memtable/bulk.rs index 6cb0feaecf..b30349cc38 100644 --- a/src/mito2/src/memtable/bulk.rs +++ b/src/mito2/src/memtable/bulk.rs @@ -40,6 +40,7 @@ use rayon::prelude::*; use serde::{Deserialize, Serialize}; use serde_with::{DisplayFromStr, serde_as}; use store_api::metadata::RegionMetadataRef; +use store_api::mito_engine_options::FloatFieldEncoding; use store_api::storage::{ColumnId, FileId, RegionId, SequenceRange}; use tokio::sync::Semaphore; @@ -435,6 +436,7 @@ pub struct BulkMemtable { merge_mode: MergeMode, /// Max number of rows in a parquet row group for encoded parts. row_group_size: usize, + float_field_encoding: FloatFieldEncoding, } impl std::fmt::Debug for BulkMemtable { @@ -665,27 +667,7 @@ impl Memtable for BulkMemtable { } fn fork(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef { - Arc::new(Self { - id, - config: self.config.clone(), - parts: Arc::new(RwLock::new(BulkParts::default())), - metadata: metadata.clone(), - alloc_tracker: AllocTracker::new(self.alloc_tracker.write_buffer_manager()), - max_timestamp: AtomicI64::new(i64::MIN), - min_timestamp: AtomicI64::new(i64::MAX), - max_sequence: AtomicU64::new(0), - num_rows: AtomicUsize::new(0), - compactor: Arc::new(Mutex::new(MemtableCompactor::new( - metadata.region_id, - id, - self.config.clone(), - self.row_group_size, - ))), - compact_dispatcher: self.compact_dispatcher.clone(), - append_mode: self.append_mode, - merge_mode: self.merge_mode, - row_group_size: self.row_group_size, - }) + self.fork_inner(id, metadata) } fn compact(&self, for_flush: bool) -> Result<()> { @@ -714,6 +696,32 @@ impl Memtable for BulkMemtable { } impl BulkMemtable { + fn fork_inner(&self, id: MemtableId, metadata: &RegionMetadataRef) -> Arc { + Arc::new(Self { + id, + config: self.config.clone(), + parts: Arc::new(RwLock::new(BulkParts::default())), + metadata: metadata.clone(), + alloc_tracker: AllocTracker::new(self.alloc_tracker.write_buffer_manager()), + max_timestamp: AtomicI64::new(i64::MIN), + min_timestamp: AtomicI64::new(i64::MAX), + max_sequence: AtomicU64::new(0), + num_rows: AtomicUsize::new(0), + compactor: Arc::new(Mutex::new(MemtableCompactor::new( + metadata.region_id, + id, + self.config.clone(), + self.row_group_size, + self.float_field_encoding, + ))), + compact_dispatcher: self.compact_dispatcher.clone(), + append_mode: self.append_mode, + merge_mode: self.merge_mode, + row_group_size: self.row_group_size, + float_field_encoding: self.float_field_encoding, + }) + } + /// Creates a new BulkMemtable with the default row group size. pub fn new( id: MemtableId, @@ -747,6 +755,32 @@ impl BulkMemtable { append_mode: bool, merge_mode: MergeMode, row_group_size: usize, + ) -> Self { + Self::new_with_row_group_size_and_encoding( + id, + config, + metadata, + write_buffer_manager, + compact_dispatcher, + append_mode, + merge_mode, + row_group_size, + FloatFieldEncoding::default(), + ) + } + + /// Creates a new BulkMemtable with the given row group size and float encoding. + #[allow(clippy::too_many_arguments)] + fn new_with_row_group_size_and_encoding( + id: MemtableId, + config: BulkMemtableConfig, + metadata: RegionMetadataRef, + write_buffer_manager: Option, + compact_dispatcher: Option>, + append_mode: bool, + merge_mode: MergeMode, + row_group_size: usize, + float_field_encoding: FloatFieldEncoding, ) -> Self { let config = config.sanitize(); let region_id = metadata.region_id; @@ -765,11 +799,13 @@ impl BulkMemtable { id, config, row_group_size, + float_field_encoding, ))), compact_dispatcher, append_mode, merge_mode, row_group_size, + float_field_encoding, } } @@ -1179,6 +1215,7 @@ struct MemtableCompactor { config: BulkMemtableConfig, /// Max number of rows in a parquet row group for encoded parts. row_group_size: usize, + float_field_encoding: FloatFieldEncoding, } impl MemtableCompactor { @@ -1188,12 +1225,14 @@ impl MemtableCompactor { memtable_id: MemtableId, config: BulkMemtableConfig, row_group_size: usize, + float_field_encoding: FloatFieldEncoding, ) -> Self { Self { region_id, memtable_id, config, row_group_size, + float_field_encoding, } } @@ -1233,6 +1272,7 @@ impl MemtableCompactor { let encode_row_threshold = self.config.encode_row_threshold; let encode_bytes_threshold = self.config.encode_bytes_threshold; let row_group_size = self.row_group_size; + let float_field_encoding = self.float_field_encoding; // Merge all groups in parallel let merged_parts = collected @@ -1247,6 +1287,7 @@ impl MemtableCompactor { encode_row_threshold, encode_bytes_threshold, row_group_size, + float_field_encoding, ) }) .collect::>>>()?; @@ -1282,6 +1323,7 @@ impl MemtableCompactor { encode_row_threshold: usize, encode_bytes_threshold: usize, row_group_size: usize, + float_field_encoding: FloatFieldEncoding, ) -> Result> { if parts_to_merge.is_empty() { return Ok(None); @@ -1382,7 +1424,11 @@ impl MemtableCompactor { if estimated_total_rows > encode_row_threshold || estimated_total_bytes > encode_bytes_threshold { - let encoder = BulkPartEncoder::new(metadata.clone(), row_group_size)?; + let encoder = BulkPartEncoder::new_with_float_field_encoding( + metadata.clone(), + row_group_size, + float_field_encoding, + )?; let mut metrics = BulkPartEncodeMetrics::default(); let encoded_part = encoder.encode_record_batch_iter( boxed_iter, @@ -1508,6 +1554,7 @@ pub struct BulkMemtableBuilder { merge_mode: MergeMode, /// Max number of rows in a parquet row group for encoded parts. row_group_size: usize, + float_field_encoding: FloatFieldEncoding, } impl Default for BulkMemtableBuilder { @@ -1519,6 +1566,7 @@ impl Default for BulkMemtableBuilder { append_mode: false, merge_mode: MergeMode::default(), row_group_size: DEFAULT_ROW_GROUP_SIZE, + float_field_encoding: FloatFieldEncoding::default(), } } } @@ -1550,6 +1598,11 @@ impl BulkMemtableBuilder { self } + pub(super) fn with_float_field_encoding(mut self, encoding: FloatFieldEncoding) -> Self { + self.float_field_encoding = encoding; + self + } + /// Sets the compact dispatcher. pub fn with_compact_dispatcher(mut self, compact_dispatcher: Arc) -> Self { self.compact_dispatcher = Some(compact_dispatcher); @@ -1564,7 +1617,7 @@ impl BulkMemtableBuilder { impl MemtableBuilder for BulkMemtableBuilder { fn build(&self, id: MemtableId, metadata: &RegionMetadataRef) -> MemtableRef { - Arc::new(BulkMemtable::new_with_row_group_size( + Arc::new(BulkMemtable::new_with_row_group_size_and_encoding( id, self.config.clone(), metadata.clone(), @@ -1573,6 +1626,7 @@ impl MemtableBuilder for BulkMemtableBuilder { self.append_mode, self.merge_mode, self.row_group_size, + self.float_field_encoding, )) } @@ -1597,6 +1651,7 @@ mod tests { use datatypes::schema::ColumnSchema; use datatypes::types::json_type::{JsonNativeType, JsonObjectType}; use mito_codec::row_converter::build_primary_key_codec; + use parquet::basic::{Encoding, Type as PhysicalType}; use serde_json::json; use store_api::metadata::{ColumnMetadata, RegionMetadataBuilder, RegionMetadataRef}; @@ -1829,6 +1884,7 @@ mod tests { usize::MAX, usize::MAX, DEFAULT_ROW_GROUP_SIZE, + FloatFieldEncoding::default(), )? .unwrap(); let MergedPart::Multi(part) = merged else { @@ -2143,7 +2199,7 @@ mod tests { #[test] fn test_bulk_memtable_fork() { let metadata = metadata_for_test(); - let original_memtable = BulkMemtable::new( + let original_memtable = BulkMemtable::new_with_row_group_size_and_encoding( 333, BulkMemtableConfig::default(), metadata.clone(), @@ -2151,6 +2207,8 @@ mod tests { None, false, MergeMode::LastRow, + DEFAULT_ROW_GROUP_SIZE, + FloatFieldEncoding::ByteStreamSplit, ); let bulk_part = @@ -2159,9 +2217,25 @@ mod tests { original_memtable.write_bulk(bulk_part).unwrap(); - let forked_memtable = original_memtable.fork(444, &metadata); + let forked_memtable = original_memtable.fork_inner(444, &metadata); assert_eq!(forked_memtable.id(), 444); + assert_eq!( + FloatFieldEncoding::ByteStreamSplit, + original_memtable.float_field_encoding + ); + assert_eq!( + original_memtable.float_field_encoding, + forked_memtable.float_field_encoding + ); + assert_eq!( + original_memtable.float_field_encoding, + forked_memtable + .compactor + .lock() + .unwrap() + .float_field_encoding + ); assert!(forked_memtable.is_empty()); assert_eq!(0, forked_memtable.stats().num_rows); @@ -2780,6 +2854,42 @@ mod tests { assert!(!BulkParts::is_merge_candidate_by_size(true, 9, 8)); } + #[test] + fn test_bulk_part_encoder_preserves_byte_stream_split() -> WhateverResult<()> { + let metadata = metadata_for_test(); + let bulk_part = create_bulk_part_with_converter( + "byte_stream_split", + 0, + vec![1000, 2000], + vec![Some(10.0), Some(20.0)], + 100, + )?; + let encoder = BulkPartEncoder::new_with_float_field_encoding( + metadata, + DEFAULT_ROW_GROUP_SIZE, + FloatFieldEncoding::ByteStreamSplit, + )?; + let encoded_part = encoder.encode_part(&bulk_part)?.unwrap(); + let field_column = encoded_part + .metadata() + .parquet_metadata + .row_groups() + .iter() + .flat_map(|row_group| row_group.columns()) + .find(|column| { + column.column_path().string() == "v1" + && column.column_type() == PhysicalType::DOUBLE + }) + .expect("Float64 field should be present in encoded bulk part"); + + assert!( + field_column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + ); + Ok(()) + } + #[test] fn test_encoded_part_batch_size_uses_largest_uncompressed_row_group() { const NUM_ROWS: usize = 14; diff --git a/src/mito2/src/memtable/bulk/part.rs b/src/mito2/src/memtable/bulk/part.rs index 2a0c42b703..f0af666a64 100644 --- a/src/mito2/src/memtable/bulk/part.rs +++ b/src/mito2/src/memtable/bulk/part.rs @@ -53,6 +53,7 @@ use smallvec::SmallVec; use snafu::{OptionExt, ResultExt}; use store_api::codec::PrimaryKeyEncoding; use store_api::metadata::{RegionMetadata, RegionMetadataRef}; +use store_api::mito_engine_options::FloatFieldEncoding; use store_api::storage::consts::PRIMARY_KEY_COLUMN_NAME; use store_api::storage::{ColumnId, FileId, SequenceNumber, SequenceRange}; @@ -70,7 +71,7 @@ use crate::sst::SeriesEstimator; use crate::sst::index::IndexOutput; use crate::sst::parquet::flat_format::primary_key_column_index; use crate::sst::parquet::format::{PrimaryKeyArray, PrimaryKeyArrayBuilder}; -use crate::sst::parquet::{PARQUET_METADATA_KEY, SstInfo}; +use crate::sst::parquet::{PARQUET_METADATA_KEY, SstInfo, apply_float_field_encoding}; const INIT_DICT_VALUE_CAPACITY: usize = 8; @@ -1306,22 +1307,29 @@ pub struct BulkPartEncoder { impl BulkPartEncoder { pub fn new(metadata: RegionMetadataRef, row_group_size: usize) -> Result { + Self::new_with_float_field_encoding(metadata, row_group_size, FloatFieldEncoding::default()) + } + + pub(super) fn new_with_float_field_encoding( + metadata: RegionMetadataRef, + row_group_size: usize, + float_field_encoding: FloatFieldEncoding, + ) -> Result { // TODO(yingwen): Skip arrow schema if needed. let json = metadata.to_json().context(InvalidMetadataSnafu)?; let key_value_meta = parquet::file::metadata::KeyValue::new(PARQUET_METADATA_KEY.to_string(), json); // TODO(yingwen): Do we need compression? - let writer_props = Some( - WriterProperties::builder() - .set_key_value_metadata(Some(vec![key_value_meta])) - .set_write_batch_size(row_group_size) - .set_max_row_group_row_count(Some(row_group_size)) - .set_compression(Compression::ZSTD(ZstdLevel::default())) - .set_column_index_truncate_length(None) - .set_statistics_truncate_length(None) - .build(), - ); + let mut props = WriterProperties::builder() + .set_key_value_metadata(Some(vec![key_value_meta])) + .set_write_batch_size(row_group_size) + .set_max_row_group_row_count(Some(row_group_size)) + .set_compression(Compression::ZSTD(ZstdLevel::default())) + .set_column_index_truncate_length(None) + .set_statistics_truncate_length(None); + props = apply_float_field_encoding(props, &metadata, float_field_encoding); + let writer_props = Some(props.build()); Ok(Self { metadata, diff --git a/src/mito2/src/region/options.rs b/src/mito2/src/region/options.rs index 9a8d17c954..a2a165c12a 100644 --- a/src/mito2/src/region/options.rs +++ b/src/mito2/src/region/options.rs @@ -32,7 +32,9 @@ use store_api::codec::PrimaryKeyEncoding; use store_api::metric_engine_consts::{ MEMTABLE_PARTITION_TREE_PRIMARY_KEY_ENCODING, PRIMARY_KEY_ENCODING, }; -use store_api::mito_engine_options::{COMPACTION_OVERRIDE, MAX_ROW_GROUP_ROW_COUNT_LIMIT}; +use store_api::mito_engine_options::{ + COMPACTION_OVERRIDE, FloatFieldEncoding, MAX_ROW_GROUP_ROW_COUNT_LIMIT, +}; use store_api::storage::{ColumnId, RegionId}; use strum::EnumString; @@ -128,6 +130,9 @@ pub struct RegionOptions { /// sequence metadata. #[serde(default, skip_serializing_if = "is_false")] pub preserve_row_sequence: bool, + /// Encoding for direct floating-point field columns in Parquet SSTs. + #[serde(default)] + pub float_field_encoding: FloatFieldEncoding, } fn is_false(value: &bool) -> bool { @@ -287,6 +292,8 @@ impl RegionOptions { sst_format = Some(FormatType::Flat); } + let float_field_encoding = options.float_field_encoding.unwrap_or_default(); + let compaction_override_flag = options_map .get(COMPACTION_OVERRIDE) .map(|v| matches!(v.to_lowercase().as_str(), "true" | "1")) @@ -322,6 +329,7 @@ impl RegionOptions { primary_key_encoding, write_buffer_size: options.write_buffer_size, preserve_row_sequence: options.preserve_row_sequence, + float_field_encoding, }; opts.validate()?; @@ -469,6 +477,9 @@ struct RegionOptionsWithoutEnum { max_row_group_row_count: Option, #[serde_as(as = "DisplayFromStr")] preserve_row_sequence: bool, + #[serde(rename = "experimental_sst_float_field_encoding")] + #[serde_as(as = "NoneAsEmptyString")] + float_field_encoding: Option, } impl Default for RegionOptionsWithoutEnum { @@ -485,6 +496,7 @@ impl Default for RegionOptionsWithoutEnum { sst_format: options.sst_format, max_row_group_row_count: options.max_row_group_row_count, preserve_row_sequence: options.preserve_row_sequence, + float_field_encoding: Some(options.float_field_encoding), } } } @@ -628,7 +640,9 @@ mod tests { use common_error::ext::ErrorExt; use common_error::status_code::StatusCode; use common_wal::options::KafkaWalOptions; - use store_api::mito_engine_options::{SKIP_WAL_KEY, WRITE_BUFFER_SIZE_KEY}; + use store_api::mito_engine_options::{ + EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, SKIP_WAL_KEY, WRITE_BUFFER_SIZE_KEY, + }; use super::*; @@ -646,6 +660,26 @@ mod tests { assert_eq!(RegionOptions::default(), options); } + #[test] + fn test_float_field_encoding_defaults_and_parses() { + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &make_map(&[])).unwrap(); + assert_eq!(FloatFieldEncoding::Default, options.float_field_encoding); + + let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "default")]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + assert_eq!(FloatFieldEncoding::Default, options.float_field_encoding); + + let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "byte_stream_split")]); + let options = RegionOptions::try_from_options(RegionId::new(0, 0), &map).unwrap(); + assert_eq!( + FloatFieldEncoding::ByteStreamSplit, + options.float_field_encoding + ); + + let map = make_map(&[(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "invalid")]); + assert!(RegionOptions::try_from_options(RegionId::new(0, 0), &map).is_err()); + } + #[test] fn test_with_ttl() { let map = make_map(&[("ttl", "7d")]); @@ -1103,6 +1137,7 @@ mod tests { primary_key_encoding: None, write_buffer_size: None, preserve_row_sequence: false, + float_field_encoding: FloatFieldEncoding::default(), }; assert_eq!(expect, options); } @@ -1161,6 +1196,7 @@ mod tests { primary_key_encoding: None, write_buffer_size: Some(ReadableSize::mb(128)), preserve_row_sequence: true, + float_field_encoding: FloatFieldEncoding::default(), }; let region_options_json_str = serde_json::to_string(&options).unwrap(); assert!(region_options_json_str.contains("preserve_row_sequence")); @@ -1172,6 +1208,7 @@ mod tests { let got: RegionOptions = serde_json::from_str(old_region_options_json_str).unwrap(); assert_eq!(None, got.write_buffer_size); assert!(!got.preserve_row_sequence); + assert_eq!(FloatFieldEncoding::Default, got.float_field_encoding); let CompactionOptions::Twcs(twcs) = got.compaction; assert_eq!(16, twcs.active_window_l1_merge_trigger); assert_eq!(8, twcs.inactive_window_l1_merge_trigger); @@ -1238,6 +1275,7 @@ mod tests { primary_key_encoding: None, write_buffer_size: None, preserve_row_sequence: false, + float_field_encoding: FloatFieldEncoding::default(), }; assert_eq!(options, got); } diff --git a/src/mito2/src/sst/parquet.rs b/src/mito2/src/sst/parquet.rs index 3d90a36092..87eeb7bae5 100644 --- a/src/mito2/src/sst/parquet.rs +++ b/src/mito2/src/sst/parquet.rs @@ -17,9 +17,14 @@ use std::collections::BTreeMap; use std::sync::Arc; +use api::v1::SemanticType; use common_base::readable_size::ReadableSize; use datatypes::json::JsonSettings; use parquet::file::metadata::ParquetMetaData; +use parquet::file::properties::WriterPropertiesBuilder; +use parquet::schema::types::ColumnPath; +use store_api::metadata::RegionMetadataRef; +use store_api::mito_engine_options::FloatFieldEncoding; use store_api::storage::{ColumnId, FileId}; use crate::sst::DEFAULT_WRITE_BUFFER_SIZE; @@ -72,6 +77,27 @@ pub(crate) struct Json2TargetLayout { /// batching without changing the row group layout of newly written SSTs. pub const DEFAULT_ROW_GROUP_SIZE: usize = 100 * 1024; +/// Applies the configured encoding to direct floating-point field columns. +pub(crate) fn apply_float_field_encoding( + mut builder: WriterPropertiesBuilder, + metadata: &RegionMetadataRef, + encoding: FloatFieldEncoding, +) -> WriterPropertiesBuilder { + if encoding == FloatFieldEncoding::ByteStreamSplit { + for column in &metadata.column_metadatas { + if column.semantic_type == SemanticType::Field + && column.column_schema.data_type.is_float() + { + let path = ColumnPath::new(vec![column.column_schema.name.clone()]); + builder = builder + .set_column_encoding(path.clone(), parquet::basic::Encoding::BYTE_STREAM_SPLIT) + .set_column_dictionary_enabled(path, false); + } + } + } + builder +} + /// Parquet write options. #[derive(Debug, Clone)] pub struct WriteOptions { @@ -83,6 +109,8 @@ pub struct WriteOptions { /// Note: This is not a hard limit as we can only observe the file size when /// ArrowWrite writes to underlying writers. pub max_file_size: Option, + /// Encoding policy for direct floating-point field columns. + pub float_field_encoding: FloatFieldEncoding, } impl Default for WriteOptions { @@ -91,6 +119,7 @@ impl Default for WriteOptions { write_buffer_size: DEFAULT_WRITE_BUFFER_SIZE, row_group_size: DEFAULT_ROW_GROUP_SIZE, max_file_size: None, + float_field_encoding: FloatFieldEncoding::default(), } } } @@ -125,6 +154,7 @@ mod tests { use std::sync::Arc; use api::v1::{OpType, SemanticType}; + use bytes::Bytes; use common_function::function::FunctionRef; use common_function::function_factory::ScalarFunctionFactory; use common_function::scalars::matches::MatchesFunction; @@ -135,20 +165,24 @@ mod tests { use datafusion_expr::{BinaryExpr, Expr, Literal, Operator, col, lit}; use datatypes::arrow; use datatypes::arrow::array::{ - ArrayRef, AsArray, BinaryDictionaryBuilder, RecordBatch, StringArray, - StringDictionaryBuilder, TimestampMillisecondArray, UInt8Array, UInt64Array, + Array, ArrayRef, AsArray, BinaryDictionaryBuilder, Float32Array, Float64Array, Int32Array, + RecordBatch, StringArray, StringDictionaryBuilder, TimestampMillisecondArray, UInt8Array, + UInt64Array, }; - use datatypes::arrow::datatypes::{DataType, Field, Schema, UInt32Type}; + use datatypes::arrow::datatypes::{DataType, Field, Schema, TimeUnit, UInt32Type}; use datatypes::arrow::util::pretty::pretty_format_batches; use datatypes::prelude::ConcreteDataType; use datatypes::schema::{FulltextAnalyzer, FulltextBackend, FulltextOptions}; use object_store::ObjectStore; - use parquet::arrow::AsyncArrowWriter; + use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + use parquet::arrow::{ArrowWriter, AsyncArrowWriter}; use parquet::basic::{Compression, Encoding, ZstdLevel}; use parquet::file::metadata::{KeyValue, PageIndexPolicy}; use parquet::file::properties::WriterProperties; + use parquet::schema::types::ColumnPath; use store_api::codec::PrimaryKeyEncoding; use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder}; + use store_api::mito_engine_options::FloatFieldEncoding; use store_api::region_request::PathType; use store_api::storage::{ColumnSchema, RegionId}; use table::predicate::Predicate; @@ -187,6 +221,176 @@ mod tests { const FILE_DIR: &str = "/"; const REGION_ID: RegionId = RegionId::new(0, 0); + #[test] + fn test_float_field_encoding_properties_and_roundtrip() { + let mut metadata_builder = RegionMetadataBuilder::new(REGION_ID); + metadata_builder + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new("f32", ConcreteDataType::float32_datatype(), true), + semantic_type: SemanticType::Field, + column_id: 0, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new("f64", ConcreteDataType::float64_datatype(), true), + semantic_type: SemanticType::Field, + column_id: 1, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new("tag", ConcreteDataType::float32_datatype(), true), + semantic_type: SemanticType::Tag, + column_id: 2, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new("i32", ConcreteDataType::int32_datatype(), true), + semantic_type: SemanticType::Field, + column_id: 3, + }) + .push_column_metadata(ColumnMetadata { + column_schema: ColumnSchema::new( + "ts", + ConcreteDataType::timestamp_millisecond_datatype(), + false, + ), + semantic_type: SemanticType::Timestamp, + column_id: 4, + }); + metadata_builder.primary_key(vec![2]); + let metadata = Arc::new(metadata_builder.build().unwrap()); + + let f32_values = [ + Some(1.5_f32), + Some(2.0), + Some(0.0), + Some(-0.0), + Some(f32::INFINITY), + Some(f32::from_bits(0x7fc0_1234)), + None, + ]; + let f64_values = [ + Some(1.5_f64), + Some(2.0), + Some(0.0), + Some(-0.0), + Some(f64::NEG_INFINITY), + Some(f64::from_bits(0x7ff8_0000_0000_1234)), + None, + ]; + let schema = Arc::new(Schema::new(vec![ + Field::new("f32", DataType::Float32, true), + Field::new("f64", DataType::Float64, true), + Field::new("tag", DataType::Float32, true), + Field::new("i32", DataType::Int32, true), + Field::new( + "ts", + DataType::Timestamp(TimeUnit::Millisecond, None), + false, + ), + ])); + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(Float32Array::from(f32_values.to_vec())) as ArrayRef, + Arc::new(Float64Array::from(f64_values.to_vec())) as ArrayRef, + Arc::new(Float32Array::from(vec![ + Some(1.0), + None, + Some(-0.0), + Some(2.0), + Some(3.0), + Some(4.0), + None, + ])), + Arc::new(Int32Array::from(vec![ + Some(1), + None, + Some(3), + Some(4), + Some(5), + Some(6), + None, + ])), + Arc::new(TimestampMillisecondArray::from_iter_values(0..7)), + ], + ) + .unwrap(); + + let path = |name: &str| ColumnPath::new(vec![name.to_string()]); + let bss = apply_float_field_encoding( + WriterProperties::builder(), + &metadata, + FloatFieldEncoding::ByteStreamSplit, + ); + assert_eq!( + Some(Encoding::BYTE_STREAM_SPLIT), + bss.clone().build().encoding(&path("f32")) + ); + assert_eq!( + Some(Encoding::BYTE_STREAM_SPLIT), + bss.clone().build().encoding(&path("f64")) + ); + assert!(!bss.clone().build().dictionary_enabled(&path("f32"))); + assert!(!bss.clone().build().dictionary_enabled(&path("f64"))); + assert_eq!(None, bss.clone().build().encoding(&path("i32"))); + assert!(bss.clone().build().dictionary_enabled(&path("tag"))); + + let default = apply_float_field_encoding( + WriterProperties::builder().set_encoding(Encoding::PLAIN), + &metadata, + FloatFieldEncoding::Default, + ) + .build(); + assert_eq!(Some(Encoding::PLAIN), default.encoding(&path("f32"))); + assert!(default.dictionary_enabled(&path("f32"))); + assert!(default.dictionary_enabled(&path("f64"))); + + let mut bytes = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut bytes, schema, Some(bss.build())).unwrap(); + writer.write(&batch).unwrap(); + let footer = writer.finish().unwrap(); + drop(writer); + for name in ["f32", "f64"] { + let column = footer.row_groups()[0] + .columns() + .iter() + .find(|column| column.column_path().string() == name) + .unwrap(); + assert!( + column + .encodings() + .any(|encoding| encoding == Encoding::BYTE_STREAM_SPLIT) + ); + } + let mut reader = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(bytes)) + .unwrap() + .build() + .unwrap(); + let actual = reader.next().unwrap().unwrap(); + let actual_f32 = actual + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let actual_f64 = actual + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for (index, value) in f32_values.into_iter().enumerate() { + assert_eq!(value.is_none(), actual_f32.is_null(index)); + if let Some(value) = value { + assert_eq!(value.to_bits(), actual_f32.value(index).to_bits()); + } + } + for (index, value) in f64_values.into_iter().enumerate() { + assert_eq!(value.is_none(), actual_f64.is_null(index)); + if let Some(value) = value { + assert_eq!(value.to_bits(), actual_f64.value(index).to_bits()); + } + } + assert!(actual.column(2).is_null(1)); + assert!(actual.column(3).is_null(1)); + } + #[derive(Clone)] struct FixedPathProvider { region_file_id: RegionFileId, diff --git a/src/mito2/src/sst/parquet/writer.rs b/src/mito2/src/sst/parquet/writer.rs index b9d5016173..9888b4ead5 100644 --- a/src/mito2/src/sst/parquet/writer.rs +++ b/src/mito2/src/sst/parquet/writer.rs @@ -38,7 +38,7 @@ use object_store::{FuturesAsyncWriter, ObjectStore}; use parquet::arrow::AsyncArrowWriter; use parquet::basic::{Compression, Encoding, ZstdLevel}; use parquet::file::metadata::KeyValue; -use parquet::file::properties::{WriterProperties, WriterPropertiesBuilder}; +use parquet::file::properties::WriterProperties; use parquet::schema::types::ColumnPath; use smallvec::smallvec; use snafu::{OptionExt, ResultExt}; @@ -61,7 +61,9 @@ use crate::sst::parquet::flat_format::{ FlatWriteFormat, primary_key_column_index, time_index_column_index, }; use crate::sst::parquet::format::{PrimaryKeyArray, PrimaryKeyWriteFormat}; -use crate::sst::parquet::{PARQUET_METADATA_KEY, SstInfo, WriteOptions}; +use crate::sst::parquet::{ + PARQUET_METADATA_KEY, SstInfo, WriteOptions, apply_float_field_encoding, +}; use crate::sst::{ DEFAULT_WRITE_BUFFER_SIZE, DEFAULT_WRITE_CONCURRENCY, FlatSchemaOptions, SeriesEstimator, maybe_wrap_schema, @@ -473,29 +475,6 @@ where Ok(results) } - /// Customizes per-column config according to schema and maybe column cardinality. - fn customize_column_config( - builder: WriterPropertiesBuilder, - region_metadata: &RegionMetadataRef, - ) -> WriterPropertiesBuilder { - let ts_col = ColumnPath::new(vec![ - region_metadata - .time_index_column() - .column_schema - .name - .clone(), - ]); - let seq_col = ColumnPath::new(vec![SEQUENCE_COLUMN_NAME.to_string()]); - let op_type_col = ColumnPath::new(vec![OP_TYPE_COLUMN_NAME.to_string()]); - - builder - .set_column_encoding(seq_col.clone(), Encoding::DELTA_BINARY_PACKED) - .set_column_dictionary_enabled(seq_col, false) - .set_column_encoding(ts_col.clone(), Encoding::DELTA_BINARY_PACKED) - .set_column_dictionary_enabled(ts_col, false) - .set_column_compression(op_type_col, Compression::UNCOMPRESSED) - } - async fn append_flat_batch( &mut self, batch: &RecordBatch, @@ -564,8 +543,22 @@ where .set_max_row_group_row_count(Some(opts.row_group_size)) .set_column_index_truncate_length(None) .set_statistics_truncate_length(None); - - let props_builder = Self::customize_column_config(props_builder, &self.metadata); + let ts_col = ColumnPath::new(vec![ + self.metadata.time_index_column().column_schema.name.clone(), + ]); + let seq_col = ColumnPath::new(vec![SEQUENCE_COLUMN_NAME.to_string()]); + let op_type_col = ColumnPath::new(vec![OP_TYPE_COLUMN_NAME.to_string()]); + let props_builder = props_builder + .set_column_encoding(seq_col.clone(), Encoding::DELTA_BINARY_PACKED) + .set_column_dictionary_enabled(seq_col, false) + .set_column_encoding(ts_col.clone(), Encoding::DELTA_BINARY_PACKED) + .set_column_dictionary_enabled(ts_col, false) + .set_column_compression(op_type_col, Compression::UNCOMPRESSED); + let props_builder = apply_float_field_encoding( + props_builder, + &self.metadata, + opts.float_field_encoding, + ); let writer_props = props_builder.build(); let sst_file_path = self.path_provider.build_sst_file_path(RegionFileId::new( diff --git a/src/store-api/src/mito_engine_options.rs b/src/store-api/src/mito_engine_options.rs index c308e91565..bc9b31ada9 100644 --- a/src/store-api/src/mito_engine_options.rs +++ b/src/store-api/src/mito_engine_options.rs @@ -88,6 +88,31 @@ pub const MAX_ROW_GROUP_ROW_COUNT: &str = "max_row_group_row_count"; pub const MAX_ROW_GROUP_ROW_COUNT_LIMIT: usize = 10 * 1024 * 1024; /// Option key for preserving per-row sequence numbers through flush and compaction. pub const PRESERVE_ROW_SEQUENCE: &str = "preserve_row_sequence"; +/// Option key for experimental Parquet float field encoding. +pub const EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING: &str = "experimental_sst_float_field_encoding"; + +/// Encoding policy for direct floating-point field columns in Parquet SSTs. +#[derive( + Debug, + Default, + Clone, + Copy, + PartialEq, + Eq, + Hash, + serde::Serialize, + serde::Deserialize, + strum::EnumString, +)] +#[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum FloatFieldEncoding { + /// The existing Parquet writer behavior. + #[default] + Default, + /// Parquet byte-stream-split encoding. + ByteStreamSplit, +} // Note: Adding new options here should also check if this option should be removed in [metric_engine::engine::create::region_options_for_metadata_region]. /// Conflicting values supplied through the legacy and canonical TWCS trigger options. @@ -155,6 +180,7 @@ pub fn is_mito_engine_option_key(key: &str) -> bool { SST_FORMAT_KEY, MAX_ROW_GROUP_ROW_COUNT, PRESERVE_ROW_SEQUENCE, + EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, ] .contains(&key) } @@ -217,9 +243,29 @@ mod tests { assert!(is_mito_engine_option_key("append_mode")); assert!(is_mito_engine_option_key("max_row_group_row_count")); assert!(is_mito_engine_option_key("preserve_row_sequence")); + assert!(is_mito_engine_option_key( + EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING + )); assert!(!is_mito_engine_option_key("foo")); } + #[test] + fn test_float_field_encoding_values() { + assert_eq!( + "default".parse::(), + Ok(FloatFieldEncoding::Default) + ); + assert_eq!( + "byte_stream_split".parse::(), + Ok(FloatFieldEncoding::ByteStreamSplit) + ); + assert_eq!( + serde_json::from_str::("\"byte_stream_split\"").unwrap(), + FloatFieldEncoding::ByteStreamSplit + ); + assert!(serde_json::from_str::("\"unknown\"").is_err()); + } + #[test] fn test_normalize_twcs_trigger_aliases_to_legacy_key() { let expected = HashMap::from([(TWCS_TRIGGER_FILE_NUM.to_string(), "4".to_string())]); diff --git a/src/table/src/requests.rs b/src/table/src/requests.rs index f7f195765f..a21a8f6fed 100644 --- a/src/table/src/requests.rs +++ b/src/table/src/requests.rs @@ -36,10 +36,10 @@ use store_api::metric_engine_consts::{ LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY, is_metric_engine_option_key, }; use store_api::mito_engine_options::{ - APPEND_MODE_KEY, COMPACTION_TYPE, MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD, - MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, MEMTABLE_BULK_MAX_MERGE_GROUPS, - MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY, SST_FORMAT_KEY, - TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, + APPEND_MODE_KEY, COMPACTION_TYPE, EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, FloatFieldEncoding, + MEMTABLE_BULK_ENCODE_BYTES_THRESHOLD, MEMTABLE_BULK_ENCODE_ROW_THRESHOLD, + MEMTABLE_BULK_MAX_MERGE_GROUPS, MEMTABLE_BULK_MERGE_THRESHOLD, MEMTABLE_TYPE, MERGE_MODE_KEY, + SST_FORMAT_KEY, TWCS_ACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_ACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_FALLBACK_TO_LOCAL, TWCS_INACTIVE_WINDOW_L1_MERGE_TRIGGER, TWCS_INACTIVE_WINDOW_TRIGGER_FILE_NUM, TWCS_MAX_OUTPUT_FILE_SIZE, TWCS_TIME_WINDOW, TWCS_TRIGGER_FILE_NUM, is_mito_engine_option_key, normalize_twcs_trigger_options, @@ -256,6 +256,16 @@ impl TableOptions { options.ttl = Some(ttl_value); } + if let Some(encoding) = kvs.get(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING) { + encoding.parse::().map_err(|_| { + ParseTableOptionSnafu { + key: EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, + value: encoding, + } + .build() + })?; + } + if let Some(skip_wal) = kvs.get(SKIP_WAL_KEY) { options.skip_wal = skip_wal.parse().map_err(|_| { ParseTableOptionSnafu { @@ -885,6 +895,7 @@ mod tests { assert!(validate_table_option(WRITE_BUFFER_SIZE_KEY)); assert!(validate_table_option(STORAGE_KEY)); assert!(validate_table_option(MEMTABLE_BULK_MERGE_THRESHOLD)); + assert!(validate_table_option(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING)); assert!(validate_table_option(REPARTITION_COLUMN_HINT_KEY)); assert!(validate_table_option(REPARTITION_PARTITION_NUM_HINT_KEY)); assert_eq!(AnnotationFamily::of_key("repartition.unknown.hint"), None); @@ -923,6 +934,25 @@ mod tests { assert!(!validate_database_option("foo")); } + #[test] + fn test_parse_float_field_encoding_option() { + let options = TableOptions::try_from_iter([( + EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, + "byte_stream_split", + )]) + .unwrap(); + assert_eq!( + options + .extra_options + .get(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING), + Some(&"byte_stream_split".to_string()) + ); + assert!( + TableOptions::try_from_iter([(EXPERIMENTAL_SST_FLOAT_FIELD_ENCODING, "invalid",)]) + .is_err() + ); + } + #[test] fn test_database_trigger_value_boundaries() { let maximum = usize::MAX.to_string(); diff --git a/tests/compatibility/cases/downgrade_compatibility/case.toml b/tests/compatibility/cases/downgrade_compatibility/case.toml index ba5ee5065e..ab0496e523 100644 --- a/tests/compatibility/cases/downgrade_compatibility/case.toml +++ b/tests/compatibility/cases/downgrade_compatibility/case.toml @@ -1,8 +1,8 @@ name = "downgrade_compatibility" -reason = "Verify v1.1.4 can reopen a table whose region WAL options were written by the current binary, and can read all rows of an append-only table flushed and compacted with preserve_row_sequence enabled." +reason = "Verify v1.1.4 can reopen tables whose region WAL options or byte-stream-split (BSS) float SSTs were written by the current binary, and can read all rows of an append-only table flushed and compacted with preserve_row_sequence enabled." introduced_by = "fix: preserve legacy region WAL options format; preserve_row_sequence" topologies = ["distributed", "standalone"] from_range = [">=v1.2.0"] to_range = ["=v1.1.4"] -features = ["table", "wal", "downgrade", "append", "preserve_row_sequence"] +features = ["table", "wal", "downgrade", "append", "preserve_row_sequence", "sst", "float", "byte_stream_split"] owner = "metasrv" diff --git a/tests/compatibility/cases/downgrade_compatibility/setup.sql b/tests/compatibility/cases/downgrade_compatibility/setup.sql index a9415570d3..267466f9df 100644 --- a/tests/compatibility/cases/downgrade_compatibility/setup.sql +++ b/tests/compatibility/cases/downgrade_compatibility/setup.sql @@ -32,3 +32,18 @@ INSERT INTO t_preserve_sequence_downgrade VALUES ADMIN FLUSH_TABLE('t_preserve_sequence_downgrade'); ADMIN COMPACT_TABLE('t_preserve_sequence_downgrade'); + +CREATE TABLE t_sst_float_field_encoding_downgrade( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + f FLOAT, + d DOUBLE +) ENGINE=mito +WITH('experimental_sst_float_field_encoding'='byte_stream_split'); + +INSERT INTO t_sst_float_field_encoding_downgrade VALUES +('2024-02-10 00:00:00+0000', 'host_a', 1.25, 10.5), +('2024-02-10 00:01:00+0000', 'host_b', -2.5, 20.25), +('2024-02-10 00:02:00+0000', 'host_c', NULL, NULL); + +ADMIN FLUSH_TABLE('t_sst_float_field_encoding_downgrade'); diff --git a/tests/compatibility/cases/downgrade_compatibility/verify.result b/tests/compatibility/cases/downgrade_compatibility/verify.result index 3c05239292..c350288d19 100644 --- a/tests/compatibility/cases/downgrade_compatibility/verify.result +++ b/tests/compatibility/cases/downgrade_compatibility/verify.result @@ -17,3 +17,15 @@ SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host; | 2024-02-09T00:02:00 | host_a | 3 | | 2024-02-09T00:03:00 | host_c | 4 | +---------------------+--------+-----+ + +SELECT ts, host, f, d +FROM t_sst_float_field_encoding_downgrade +ORDER BY ts, host; + ++---------------------+--------+------+-------+ +| ts | host | f | d | ++---------------------+--------+------+-------+ +| 2024-02-10T00:00:00 | host_a | 1.25 | 10.5 | +| 2024-02-10T00:01:00 | host_b | -2.5 | 20.25 | +| 2024-02-10T00:02:00 | host_c | | | ++---------------------+--------+------+-------+ diff --git a/tests/compatibility/cases/downgrade_compatibility/verify.sql b/tests/compatibility/cases/downgrade_compatibility/verify.sql index 764968bf1b..6b73c88611 100644 --- a/tests/compatibility/cases/downgrade_compatibility/verify.sql +++ b/tests/compatibility/cases/downgrade_compatibility/verify.sql @@ -1,3 +1,7 @@ SELECT ts, host, val FROM t_downgrade_compatibility ORDER BY ts, host; SELECT ts, host, val FROM t_preserve_sequence_downgrade ORDER BY ts, host; + +SELECT ts, host, f, d +FROM t_sst_float_field_encoding_downgrade +ORDER BY ts, host; diff --git a/tests/compatibility/cases/sst_float_field_encoding_upgrade/case.toml b/tests/compatibility/cases/sst_float_field_encoding_upgrade/case.toml new file mode 100644 index 0000000000..4f008775bb --- /dev/null +++ b/tests/compatibility/cases/sst_float_field_encoding_upgrade/case.toml @@ -0,0 +1,8 @@ +name = "sst_float_field_encoding_upgrade" +reason = "Verify newer binaries can read default-encoded Mito SSTs containing FLOAT and DOUBLE fields written by older binaries." +introduced_by = "experimental_sst_float_field_encoding" +topologies = ["distributed", "standalone"] +from_range = ["*"] +to_range = ["*"] +features = ["table", "sst", "float", "upgrade"] +owner = "storage" diff --git a/tests/compatibility/cases/sst_float_field_encoding_upgrade/setup.sql b/tests/compatibility/cases/sst_float_field_encoding_upgrade/setup.sql new file mode 100644 index 0000000000..a84b4339df --- /dev/null +++ b/tests/compatibility/cases/sst_float_field_encoding_upgrade/setup.sql @@ -0,0 +1,13 @@ +CREATE TABLE t_sst_float_field_encoding_upgrade( + ts TIMESTAMP TIME INDEX, + host STRING PRIMARY KEY, + f FLOAT, + d DOUBLE +) ENGINE=mito; + +INSERT INTO t_sst_float_field_encoding_upgrade VALUES +('2024-02-10 00:00:00+0000', 'host_a', 1.25, 10.5), +('2024-02-10 00:01:00+0000', 'host_b', -2.5, 20.25), +('2024-02-10 00:02:00+0000', 'host_c', NULL, NULL); + +ADMIN FLUSH_TABLE('t_sst_float_field_encoding_upgrade'); diff --git a/tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.result b/tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.result new file mode 100644 index 0000000000..d6ae25d974 --- /dev/null +++ b/tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.result @@ -0,0 +1,11 @@ +SELECT ts, host, f, d +FROM t_sst_float_field_encoding_upgrade +ORDER BY ts, host; + ++---------------------+--------+------+-------+ +| ts | host | f | d | ++---------------------+--------+------+-------+ +| 2024-02-10T00:00:00 | host_a | 1.25 | 10.5 | +| 2024-02-10T00:01:00 | host_b | -2.5 | 20.25 | +| 2024-02-10T00:02:00 | host_c | | | ++---------------------+--------+------+-------+ diff --git a/tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.sql b/tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.sql new file mode 100644 index 0000000000..81822cb0ac --- /dev/null +++ b/tests/compatibility/cases/sst_float_field_encoding_upgrade/verify.sql @@ -0,0 +1,3 @@ +SELECT ts, host, f, d +FROM t_sst_float_field_encoding_upgrade +ORDER BY ts, host; diff --git a/tests/perf/README.md b/tests/perf/README.md index 6f7893ae66..0361f598a3 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -86,15 +86,19 @@ sample values without changing label cardinality or the runner lifecycle: [scenario.remote_write.value] pattern = "quantized_signal" # linear, constant, modulo, unique, seeded_random, # run_length, quantized_signal, - # signal_with_sporadic_stalls, mixed_signal_repeated + # signal_with_sporadic_stalls, mixed_signal_repeated, + # bounded_mixed base = 0.0 step = 0.125 -cardinality = 4096 # buckets for modulo/seeded_random/quantized_signal/run_length -seed = 12345 # deterministic seeded_random input -run_length = 8 # adjacent samples per bucket for run_length/quantized_signal +cardinality = 4096 # buckets for modulo/seeded_random/quantized_signal/run_length; + # baseline buckets for bounded_mixed +seed = 12345 # deterministic seeded_random input and bounded_mixed series hash +run_length = 8 # adjacent samples per bucket for run_length/quantized_signal; + # interpolation interval for bounded_mixed stall_every = 100 # interval for signal_with_sporadic_stalls stall_length = 16 # held samples inside each stall interval -mixed_every = 5 # every Nth sample becomes the repeated base value +mixed_every = 5 # every Nth sample becomes the repeated base value; + # one fractional bounded_mixed series per N series ``` The default `linear` pattern preserves the helper's historical formula. Use @@ -103,12 +107,29 @@ data shapes, `run_length` for run-heavy low-cardinality series, `quantized_signa for signal-like values collapsed into a finite bucket set, `signal_with_sporadic_stalls` for mostly continuous signals with periodic flat spots, and `mixed_signal_repeated` for signal-plus-periodic-default mixtures. `unique` or high-cardinality buckets -still work for broad sample-value distributions. This is a generic sample-value -control for query/ingestion cases; it does not inspect or assert storage -encoding, Parquet footers, or storage policy choices. For chunked remote-write -ingestion, the runner passes the sample offset and total sample count to the -helper so non-linear value patterns use a stable global/per-series ordinal across -chunks. +still work for broad sample-value distributions. + +`bounded_mixed` is synthetic rather than an empirical workload. For series `s`, +`h = splitmix64(s ^ seed)` selects a span from `[10, 1_000, 100_000, 10_000_000]` +using its top two bits, multiplied by `1 + ((h >> 32) % 10)`. Its baseline is +`(h % max(cardinality, 1)) * span / 10`. Ranges can overlap. At local sample +`l = sample_offset + sample_idx`, it linearly interpolates hash-derived anchors +in `[-span + 1, span - 1]`, changing anchors every `max(run_length, 1)` samples. +It applies `base + step * signal`, then rounds to an integer. Every +`max(mixed_every, 1)`-th series adds a hash-derived fraction from `1/1000` to +`999/1000`. The 95/5 finite, nonintegral guarantee applies to the case's bounded +parameters, not arbitrary huge or nonfinite base/step inputs. With `base = 0` +and `step = 1`, output lies within `baseline ± span`, allowing one unit for +rounding and the fractional addition. The explicit case uses 1,000 equal-length +series and `mixed_every = 20`, giving exactly 95% integer-valued and 5% +fractional Float64 samples. Its ranges and temporal shape are synthetic design +parameters, not measured properties from the survey. + +This is a generic sample-value control for query/ingestion cases; it does not +inspect or assert storage encoding, Parquet footers, or storage policy choices. +For chunked remote-write ingestion, the runner passes the sample offset and total +sample count to the helper so non-linear value patterns use a stable global/per-series +ordinal across chunks. Case schema, value distribution defaults, storage defaults, and read-bench defaults are owned by Rust. The outer CI driver calls @@ -162,12 +183,15 @@ measures region scan cost, and query measurements still exercise the SQL/TQL frontend path. Treat all performance conclusions as release-only; debug builds are suitable only for command wiring and correctness checks. -`prepare-remote` creates the configured database if needed. The outer driver writes a per-target -frontend config enabling `[prom_store]` with metric engine storage and a non-zero -`pending_rows_flush_interval`, and validates that the logical metric table reaches -`series_count * samples_per_series` rows before trusting the query measurements. -Use `--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper -to the outer driver. +`prepare-remote` creates the configured database if needed. A remote-write case +can provide `base_setup_sql` and `candidate_setup_sql` lists; each target runs its +own complete statements in order after database creation and before ingestion. +The outer driver writes a per-target frontend config enabling `[prom_store]` with +metric engine storage and a non-zero `pending_rows_flush_interval`, and validates +that the logical metric table reaches `series_count * samples_per_series` rows +before trusting the query measurements. Use +`--fixture-generator /path/to/query_perf_fixture` to provide the Rust helper to +the outer driver. Large manual remote-write cases can set `sample_chunk_size` to split ingestion by time. For each chunk, `prepare-remote` invokes `query_perf_fixture prom-remote-write` with the @@ -204,7 +228,9 @@ case for issue #7913. It writes 8192 series × 20160 samples through remote-writ in 1440-sample daily time chunks, flushing after each chunk before running 1d/7d/14d TQL selectors. It is not included in the default `all` case set because ingestion cost dominates routine CI validation. Commenting `/query-regression heavy` runs -only this case; `/query-regression` runs the eight routine default cases. Manual +only this case; `/query-regression` runs the nine routine default cases, including +`sst_float_bss`, `promql_instant_last_row_9034`, and +`mito_prefilter_all_match`. Manual workflow dispatch accepts the `heavy` token to select this case. The routine default set also includes @@ -371,6 +397,89 @@ uv run --no-project python .github/scripts/query-regression-run.py \ --work-dir /tmp/query-regression-work ``` +### SST float BSS comparison + +`tests/perf/query_cases/sst_float_bss/case.toml` is included in the routine +`all` default case group and compares a default empty physical metric table with +a candidate byte-stream-split (BSS) physical table. It writes 1,000 series × 4,320 samples +(4,320,000 rows) using synthetic `bounded_mixed` values: 95% integral series and +5% nonintegral series, with bounded per-series fluctuation rather than globally +unique values. This deliberately matches a 95/5 design; it is not an empirical +claim about any production population. A separate survey sample found 94.79% +integral values, but that observation does not make its values globally unique. +Current mixed-data evidence, including SST inspection, exact-bit row verification, +warmed endpoint SQL, and both warm-reader projections, is recorded in +[`query_cases/sst_float_bss/RESULTS.md`](query_cases/sst_float_bss/RESULTS.md). +Timing observations were collected while other builds saturated the shared host; +they are not performance acceptance evidence. Before rerunning timings, ensure +the host is idle (not merely this agent), record load throughout, and avoid +concurrent builds. CPU affinity alone does not isolate memory or I/O contention. +The historical unique-integer workload is preserved only in Git history and does +not apply to this case. +Both targets must use the exact same release `greptime` binary; only the +per-target table setup SQL differs. Run the existing driver from a checkout +containing that binary, with absolute paths and fresh data directories for every +run: + +```bash +REPO="$(pwd -P)" +RELEASE_GREPTIME="/absolute/path/to/release/greptime" +FIXTURE_GENERATOR="/absolute/path/to/release/query_perf_fixture" +RUNNER="/absolute/path/to/release/query_regression_runner" +WORK_DIR="/absolute/path/to/fresh/sst-float-bss-run-1" +cd "$REPO" +uv run --no-project python "$REPO/.github/scripts/query-regression-run.py" \ + --cases "$REPO/tests/perf/query_cases/sst_float_bss/case.toml" \ + --base-src "$REPO" \ + --candidate-src . \ + --base-bin "$RELEASE_GREPTIME" \ + --candidate-bin "$RELEASE_GREPTIME" \ + --fixture-generator "$FIXTURE_GENERATOR" \ + --runner "$RUNNER" \ + --work-dir "$WORK_DIR" \ + --summary-script "$REPO/.github/scripts/query-regression-summary.py" +``` + +Repeat the command three times with a different fresh absolute `WORK_DIR` each +run. In each `query-regression-report.json`, compare base and candidate +`targets[].storage_inspection.summary.summary.total_file_size`, and each query's +`targets[].measurements[].latency_ms_median`. Storage percentage is +`(candidate_total_file_size - base_total_file_size) / base_total_file_size * 100`; +query latency percentage is +`(candidate_latency_ms_median - base_latency_ms_median) / base_latency_ms_median * 100`. +The configured three warmups occur after the initial query validation and before +that query's 15 measured endpoint requests, so query latency is a warmed +frontend/cache measurement. The case restricts storage inspection to +`data/greptime/public`, excluding `greptime_private` SSTs. After the datanodes +stop, it also runs seven iterations of value-only `parquetbench` for all +inspected SST files and sequential `scanbench` over the corresponding regions, +with parallelism one. The three explicit flushes yielded six SSTs in the observed +mixed runs—three 1,105,920-row files and three 334,080-row files—because storage split each +flush; this is an observation, not a guaranteed file layout. Their per-run +output and aggregate `parquetbench_median_average_ms` and +`scanbench_median_average_ms` are under +`targets[].read_bench`; they are quiescent local-file read/scan diagnostics, not +warmed frontend-query latency measurements. Bench averages include iteration one; +no OS cache is dropped, and the driver runs base before candidate. + +Historical [PR #8548](https://github.com/GreptimeTeam/greptimedb/pull/8548) +reported storage savings alongside warm-read slowdowns for a different mixed +counter/gauge study. It used value-only and all-column projections, discarded the +first iteration, and alternated target order. This case reuses that reader +measurement approach, not its dataset or results. For the supplementary warm +comparison, reuse recorded commands against stopped data directories with eight +iterations, discard iteration one, and alternate target order for eight rounds. +For parquetbench, sum all per-file warm medians before taking the outer median; +for scanbench, take the outer median of whole-region warm medians. Keep +post-flush and post-compaction measurements separate. + +The case's `-5.0` storage target and `25` query-latency guardrail are experimental +acceptance targets, not observed-benefit claims; do not relax them if a run +fails. Its three periodic flushes and shared high TWCS trigger avoid the normal +four-file compaction trigger from confounding the layout. Check footer encodings +(BSS on candidate and no BSS on base) and data equality from the artifacts rather +than through a new harness framework. + For a focused manual reproduction of the Mito prefilter all-match optimization, use the existing lifecycle command above with `--cases tests/perf/query_cases/mito_prefilter_all_match/case.toml`. Its four count probes require explicit result inspection rather than automatic validation: expect `262144`, `131072`, `16896`, and `16896` in query order. The default lifecycle retains caches, so the optimization remains exercised but its timing includes warm-cache effects; use an explicitly configured cold environment when a cold comparison is required. The Rust runner subcommands are also useful for focused diagnostics: @@ -400,8 +509,8 @@ parquetbench/scanbench` as the read-bench tool against each target's data direct The workflow runs when an allowlisted repository admin comments `/query-regression` on a non-draft PR. It does not rerun on pushes, -ready-for-review, or reopen events. `/query-regression` runs the eight routine -default cases, including `promql_instant_last_row_9034` and +ready-for-review, or reopen events. `/query-regression` runs the nine routine +default cases, including `sst_float_bss`, `promql_instant_last_row_9034`, and `mito_prefilter_all_match`; `/query-regression heavy` runs only the high-cardinality remote-write #7913 case. PR runs build base/candidate once and use `--allow-large-fixture`. Manual `workflow_dispatch` runs can pass `all`, diff --git a/tests/perf/query_cases/sst_float_bss/RESULTS.md b/tests/perf/query_cases/sst_float_bss/RESULTS.md new file mode 100644 index 0000000000..ba932d0e42 --- /dev/null +++ b/tests/perf/query_cases/sst_float_bss/RESULTS.md @@ -0,0 +1,145 @@ +# SST float BSS: mixed-data post-flush evidence + +This report replaces the earlier unique-integer experiment as the primary evidence +for `sst_float_bss`. The detailed historical numbers remain in Git history; they +apply only to that superseded workload and must not be used for the current mixed +case. + +> **Timing measurements are confounded.** After these runs, the user reported +> that other agents' builds saturated all 20 logical CPUs during measurement. +> No host-wide idle/load gate was enforced. The latency tables below are retained +> as raw observations only: neither improvements nor threshold failures establish +> BSS performance. SST byte counts and cross-target stored-row checks remain +> valid for the inspected dataset. Stable-host timing verification is pending. + +## Case and data shape + +- Generator/case revision: `149b49de5f3`. +- Base and candidate used the same release `greptime` binary. The only physical + table difference was `experimental_sst_float_field_encoding`: `default` versus + `byte_stream_split`; base used dictionary encoding and candidate used BSS + without dictionaries. This is not a dictionary-policy-held-constant comparison. +- Each of three fresh post-flush runs wrote 1,000 series × 4,320 samples at + 60-second intervals: 4,320,000 rows per target. Three explicit flushes produced + six SSTs per target: three × 1,105,920 rows and three × 334,080 rows, with 45 + `greptime_value` row groups total. This is observed layout, not a layout + guarantee. +- `bounded_mixed` is a synthetic, uncalibrated design: a series hash selects its + baseline and span band, and 60-sample anchor interpolation produces bounded + fluctuation. It does not represent a measured production distribution. +- Exactly 950 equal-length series are integer-only and 50 are fractional-only, + yielding 4,104,000 integer rows and 216,000 fractional rows per target. The + equal-length 1,000-series design deliberately makes both the series and row + shares 95%/5%. +- The 95% design is not an empirical claim. A [local preview survey project](http://192.168.50.85:8765/view/prometheus-metrics-186-project-preliminary) + observed 94.79% integral **samples**; it does not establish a global rate, + global uniqueness, or a 95% series distribution. + +## Initial SST inspection and value verification + +Inspection is limited to `data/greptime/public`; bytes exclude WAL, indexes, +metadata, and internal/private-table storage. Each of the three runs had the +same totals: + +| Measure | Default | BSS | BSS change | +| --- | ---: | ---: | ---: | +| Selected SST bytes | 20,443,131 | 8,884,743 | -56.5392% | +| `greptime_value` compressed bytes | 20,201,188 | 8,643,964 | -57.2106% | +| SST files / rows / value row groups | 6 / 4,320,000 / 45 | 6 / 4,320,000 / 45 | — | + +Footer inspection found `PLAIN`/`RLE`/`RLE_DICTIONARY` for the default value +column and `RLE`/`BYTE_STREAM_SPLIT` for BSS (both ZSTD level 1). The configured +storage threshold remains a 5% reduction; this repeated SST-size result passes +that threshold. It is post-flush evidence only, not a compaction result or a +general float-compression claim. + +For every base/candidate group in all three runs, rows were compared by the same +primary key and timestamp and their Float64 bits were exact matches: 950 +integer-only and 50 fractional-only series, 4,104,000 and 216,000 rows +respectively. The six canonical sorted-row SHA-256 results are identical: + +``` +48f87d7d01bc8a0ce9a9cdb3033f322e787784fcae61f1417ac25b05c4a40375 +``` + +This verifies exact cross-target stored rows, not independent reconstruction of +the generator output. All 1,000 series had both rises and falls, and every one +of the 36 inspected SSTs contained both integer-valued and fractional values; +these were not stored as two disjoint file sets. + +## Warmed endpoint SQL + +Each query used three warmups followed by 15 measured endpoint requests. Values +are median/p95 milliseconds; delta is `(BSS median - default median) / default +median × 100%`. The guardrail remains a maximum 25% candidate median regression and +was not changed for these runs. + +| Run | Query | Default median / p95 | BSS median / p95 | Delta | Guardrail | +| --- | --- | ---: | ---: | ---: | --- | +| mixed-1 | `sum_all_values` | 27.13 / 37.38 | 25.45 / 38.99 | -6.21% | pass | +| mixed-1 | `hourly_sum_values` | 118.78 / 136.63 | 130.93 / 185.26 | +10.23% | pass | +| mixed-2 | `sum_all_values` | 24.02 / 35.02 | 36.34 / 56.01 | +51.31% | **fail** | +| mixed-2 | `hourly_sum_values` | 95.58 / 120.40 | 85.45 / 148.03 | -10.60% | pass | +| mixed-3 | `sum_all_values` | 38.25 / 55.41 | 19.68 / 29.49 | -48.55% | pass | +| mixed-3 | `hourly_sum_values` | 56.77 / 148.29 | 121.87 / 143.16 | +114.66% | **fail** | + +The two recorded threshold failures remain in the artifacts, but the saturated +shared host prevents attributing them to BSS. Passing entries likewise do not +establish performance acceptance. No CPU isolation or cache flush was used. + +## Stopped-directory warm readers + +For each of the three datasets, eight outer rounds alternate base/candidate order. +Each command runs eight iterations, discards the first, and takes the median of +iterations 2–8. Parquetbench reads every SST: sum the six per-file medians, then +take the median of the eight round totals. Scanbench scans the whole region and +reports the outer median of its eight warm medians. These are distinct metrics, +not interchangeable whole-query latencies. Deltas are ratios of separately +aggregated medians. Value projection selects `greptime_value`; all-column +parquetbench returns five physical columns, while scanbench uses `{}` (all +region columns). Projection order is fixed; target order alternates. + +| Dataset | Projection | Reader | Base ms | BSS ms | Change | +| --- | --- | --- | ---: | ---: | ---: | +| mixed-1 | value | parquetbench | 95.957 | 68.442 | -28.67% | +| mixed-1 | value | scanbench | 144.293 | 115.421 | -20.01% | +| mixed-1 | allcolumns | parquetbench | 176.620 | 103.761 | -41.25% | +| mixed-1 | allcolumns | scanbench | 200.881 | 202.660 | +0.89% | +| mixed-2 | value | parquetbench | 112.151 | 74.275 | -33.77% | +| mixed-2 | value | scanbench | 109.824 | 78.189 | -28.80% | +| mixed-2 | allcolumns | parquetbench | 141.944 | 102.681 | -27.66% | +| mixed-2 | allcolumns | scanbench | 211.414 | 163.261 | -22.78% | +| mixed-3 | value | parquetbench | 121.827 | 67.565 | -44.54% | +| mixed-3 | value | scanbench | 100.968 | 79.878 | -20.89% | +| mixed-3 | allcolumns | parquetbench | 113.984 | 94.361 | -17.22% | +| mixed-3 | allcolumns | scanbench | 193.101 | 164.175 | -14.98% | + +All 672 commands / 5,376 raw iterations returned the expected rows; timing, +projection configurations, parquet schemas and order metadata were checked. +Run 1 all-column scanbench differed by **+0.89%**, negligible relative to the +observed variation, not evidence of a regression. Its round medians ranged from +167.197–663.465 ms (base) and 133.339–557.352 ms (BSS). There is substantial +variability; these measurements do not establish a stable gain on every read +path. The reported concurrent CPU saturation affects apparent improvements as +well as slowdowns; no causal performance conclusion is drawn from these runs. + +## Scope and recorded checks + +This case compares default and BSS post-flush data only. It adds no benchmark +framework and no public artifact upload. Historical unique-integer results and +other mixed counter/gauge studies have different data and methods; they are +context only and do not apply to this workload. + +Recorded checks for this change set: 37 targeted Rust tests, 20 Python tooling +tests, and 15 plans passed. Full-workspace tests were not run. + +Local artifacts are under +`/mnt/nvme_rust/rust-targets/metric-bss-perf/experiments/mixed-{1,2,3}/`, +including `value-verification.json`, `verification.txt`, and each run's +`sst_float_bss/query-regression-report.json`. + +Warm artifacts are in sibling `warm-mixed-{1,2,3}/` directories; +`mixed-warm-comparison.json` contains the table and per-round ranges. +`mixed-file-composition.json` records integer/fractional counts for each SST. +The shared release `greptime` SHA-256 is +`2ce0ab1670cd4bdb0f60c87af3e1f494ee66c3c6ab83bce943a073638514eb2b`. diff --git a/tests/perf/query_cases/sst_float_bss/case.toml b/tests/perf/query_cases/sst_float_bss/case.toml new file mode 100644 index 0000000000..272e0eb4bc --- /dev/null +++ b/tests/perf/query_cases/sst_float_bss/case.toml @@ -0,0 +1,83 @@ +# Default all-group default-versus-BSS storage and warmed-query comparison. + +[case] +name = "sst_float_bss" +description = "Compare default and byte-stream-split floating-point SST bytes and warmed value-scanning query latency with bounded 95/5 per-series DOUBLE values" + +[scenario] +kind = "prom_remote_write_then_query" + +[scenario.remote_write] +database = "public" +metric = "sst_float_bss" +physical_table = "sst_float_bss_physical" +series_count = 1000 +samples_per_series = 4320 +sample_chunk_size = 1440 +flush_every_sample_chunks = 1 +start_unix_millis = 1_704_067_200_000 # 2024-01-01T00:00:00Z +step_millis = 60000 +chunk_series_count = 128 +timeout_seconds = 600 +visibility_timeout_seconds = 300 + +# Omit the default encoding from base setup because older binaries do not recognize it. +# The high shared TWCS trigger avoids compaction changing the three-flush layout. +base_setup_sql = [ + "CREATE TABLE sst_float_bss_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'compaction.twcs.trigger_file_num'='100')", +] +candidate_setup_sql = [ + "CREATE TABLE sst_float_bss_physical (greptime_timestamp TIMESTAMP TIME INDEX, greptime_value DOUBLE) ENGINE=metric WITH ('physical_metric_table'='', 'experimental_sst_float_field_encoding'='byte_stream_split', 'compaction.twcs.trigger_file_num'='100')", +] + +[scenario.remote_write.value] +# Synthetic, not empirical: 95% integral and 5% nonintegral series vary +# smoothly within deterministic per-series ranges; series are not a global monotonic counter. +pattern = "bounded_mixed" +base = 0 +step = 1 +cardinality = 1000 +seed = 42 +mixed_every = 20 +run_length = 60 + +[scenario.remote_write.prom_store] +pending_rows_flush_interval = "1s" +max_batch_rows = 100000 + +[scenario.remote_write.storage] +root_suffix = "data/greptime/public" +column = "greptime_value" +min_files = 1 +min_files_with_column = 1 +max_candidate_total_file_size_regression_pct = -5.0 + +[scenario.remote_write.read_bench] +enabled = true +parquetbench = true +scanbench = true +iterations = 7 +projection = ["greptime_value"] +parquet_reader = "direct" +scan_scanner = "seq" +parallelism = 1 + +[[scenario.queries]] +name = "sum_all_values" +kind = "sql" +query = "SELECT sum(greptime_value) FROM sst_float_bss" +warmup = 3 +iterations = 15 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 25 + +[[scenario.queries]] +name = "hourly_sum_values" +kind = "sql" +query = "SELECT date_bin(INTERVAL '1 hour', greptime_timestamp) AS time_window, sum(greptime_value) FROM sst_float_bss WHERE greptime_timestamp >= TIMESTAMP '2024-01-01 00:00:00' AND greptime_timestamp < TIMESTAMP '2024-01-04 00:00:00' GROUP BY time_window ORDER BY time_window" +warmup = 3 +iterations = 15 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 25 diff --git a/tests/perf/test_query_regression_case_selection.py b/tests/perf/test_query_regression_case_selection.py index 6ccf18a442..04a542e23a 100644 --- a/tests/perf/test_query_regression_case_selection.py +++ b/tests/perf/test_query_regression_case_selection.py @@ -16,6 +16,7 @@ """Coverage for query-regression case group selection.""" import importlib.util +import re import sys import unittest from pathlib import Path @@ -35,6 +36,7 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase): runner.split_cases(["all"]), [ "tests/perf/query_cases/smoke_direct_sst/case.toml", + "tests/perf/query_cases/sst_float_bss/case.toml", "tests/perf/query_cases/prom_remote_write_seeded_random/case.toml", "tests/perf/query_cases/prom_remote_write_run_heavy/case.toml", "tests/perf/query_cases/prom_remote_write_mixed_every/case.toml", @@ -45,8 +47,8 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase): ], ) - def test_implicit_selection_has_eight_routine_cases(self) -> None: - self.assertEqual(len(runner.DEFAULT_CASES), 8) + def test_implicit_selection_has_nine_routine_cases(self) -> None: + self.assertEqual(len(runner.DEFAULT_CASES), 9) self.assertEqual(runner.split_cases([]), runner.DEFAULT_CASES) def test_heavy_selects_only_remote_write_7913(self) -> None: @@ -59,6 +61,23 @@ class QueryRegressionCaseSelectionTest(unittest.TestCase): case = "tests/perf/query_cases/sql_topk_order_by/case.toml" self.assertEqual(runner.split_cases([case]), [case]) + def test_sst_float_bss_omits_base_encoding_for_older_binaries(self) -> None: + case_path = Path(__file__).parent / "query_cases/sst_float_bss/case.toml" + setup_sql = dict(re.findall( + r'(base|candidate)_setup_sql = \[\s*"([^"\n]+)"', + case_path.read_text(), + )) + base_setup = setup_sql["base"] + candidate_setup = setup_sql["candidate"] + candidate_option = "'experimental_sst_float_field_encoding'='byte_stream_split'" + + self.assertNotIn("experimental_sst_float_field_encoding", base_setup) + self.assertIn(candidate_option, candidate_setup) + self.assertEqual( + base_setup, + candidate_setup.replace(f", {candidate_option}", ""), + ) + if __name__ == "__main__": unittest.main()