From e12fd660fc6bb70194f5e45e759df77882fd342f Mon Sep 17 00:00:00 2001 From: discord9 <55937128+discord9@users.noreply.github.com> Date: Tue, 15 Sep 2026 12:17:12 +0800 Subject: [PATCH] fix(promql): address constant-tag review and cardinality coverage Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --- .../src/extension_plan/series_divide.rs | 51 ++++++- .../promql_constant_tag_concat_ms/case.toml | 1 - .../case.toml | 138 ++++++++++++++++++ .../case.toml | 34 ++--- 4 files changed, 205 insertions(+), 19 deletions(-) create mode 100644 tests/perf/query_cases/promql_constant_tag_concat_ms_100k/case.toml rename tests/perf/query_cases/{promql_constant_tag_concat => promql_constant_tag_concat_ms_10k}/case.toml (81%) diff --git a/src/promql/src/extension_plan/series_divide.rs b/src/promql/src/extension_plan/series_divide.rs index d9be6adfbd4..3bf8a448205 100644 --- a/src/promql/src/extension_plan/series_divide.rs +++ b/src/promql/src/extension_plan/series_divide.rs @@ -521,13 +521,32 @@ fn concat_series_batches( batches: &[RecordBatch], tag_indices: &[usize], ) -> DataFusionResult { - if batches.len() <= 1 || tag_indices.is_empty() || schema.fields().is_empty() { + if batches.len() <= 1 || tag_indices.is_empty() { return Ok(compute::concat_batches(schema, batches)?); } let Some(first_batch) = batches.iter().find(|batch| batch.num_rows() > 0) else { return Ok(compute::concat_batches(schema, batches)?); }; + + // This endpoint-only sanity check does not validate tags in interior rows. + #[cfg(debug_assertions)] + { + let last_batch = batches + .iter() + .rfind(|batch| batch.num_rows() > 0) + .expect("first non-empty batch implies a last non-empty batch"); + if let (Ok(first_tags), Ok(last_tags)) = ( + TagIdentifier::try_new(first_batch, tag_indices), + TagIdentifier::try_new(last_batch, tag_indices), + ) { + debug_assert!( + first_tags.equal_at(0, &last_tags, last_batch.num_rows() - 1)?, + "series batch tag endpoints must match" + ); + } + } + let total_rows: usize = batches.iter().map(RecordBatch::num_rows).sum(); let mut take_indices = None; let columns = schema @@ -773,6 +792,36 @@ mod test { assert_concat_matches_reference(schema, batches, &[0]); } + #[cfg(debug_assertions)] + #[test] + #[should_panic(expected = "series batch tag endpoints must match")] + fn test_concat_series_batches_mismatched_tag_endpoints_panics() { + let schema = Arc::new(Schema::new(vec![ + Field::new("tag", DataType::Utf8, true), + Field::new("value", DataType::Int64, true), + ])); + let batches = vec![ + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["first"])), + Arc::new(Int64Array::from(vec![1])), + ], + ) + .unwrap(), + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(StringArray::from(vec!["last"])), + Arc::new(Int64Array::from(vec![2])), + ], + ) + .unwrap(), + ]; + + concat_series_batches(&schema, &batches, &[0]).unwrap(); + } + #[test] fn test_concat_series_batches_string_tags() { for (data_type, batches) in [ diff --git a/tests/perf/query_cases/promql_constant_tag_concat_ms/case.toml b/tests/perf/query_cases/promql_constant_tag_concat_ms/case.toml index 0a4f7114753..66e432d90f7 100644 --- a/tests/perf/query_cases/promql_constant_tag_concat_ms/case.toml +++ b/tests/perf/query_cases/promql_constant_tag_concat_ms/case.toml @@ -1,5 +1,4 @@ # Millisecond timestamps exercise PerSeries/SeriesScan without the ns-to-ms cast. -# Keep the nanosecond case as the repartition-and-sort control. # CI performance qualification for the constant-tag concat experiment. # # This direct-SST fixture fixes file, row-group, and label layout independently diff --git a/tests/perf/query_cases/promql_constant_tag_concat_ms_100k/case.toml b/tests/perf/query_cases/promql_constant_tag_concat_ms_100k/case.toml new file mode 100644 index 00000000000..59e45f226bc --- /dev/null +++ b/tests/perf/query_cases/promql_constant_tag_concat_ms_100k/case.toml @@ -0,0 +1,138 @@ +# Millisecond timestamp high-cardinality qualification for constant-tag concat. +# +# This direct-SST fixture fixes file, row-group, and label layout independently +# of ingestion and SQL flush behavior; it does not claim to reproduce a +# SQL-flush baseline. With timestamp-major generation, every series occurs at +# every scrape timestamp and each series has eight samples in all 32 +# non-overlapping SSTs, exercising cross-file reads for the multi-evaluation queries. + +[case] +name = "promql_constant_tag_concat_ms_100k" +description = "PromQL selector and topk/bottomk qualification on a controlled constant-tag direct-SST fixture" + +[scenario] +kind = "direct_readable_sst" +seed = 4096 + +[[scenario.tables]] +database = "public" +name = "promql_constant_tag_concat_ms_100k" +engine = "mito" +append_mode = true +sst_format = "flat" +primary_key = ["host", "instance"] +time_index = "ts" + +[[scenario.tables.columns]] +name = "host" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 16, prefix = "host" } + +[[scenario.tables.columns]] +name = "instance" +type = "STRING" +semantic = "tag" +distribution = { kind = "cardinality", values = 100_000, prefix = "instance" } + +[[scenario.tables.columns]] +name = "reading" +type = "DOUBLE" +semantic = "field" +distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 } + +[[scenario.tables.columns]] +name = "ts" +type = "TIMESTAMP(3)" +semantic = "timestamp" + +[scenario.layout] +regions = 1 +sst_count = 32 +rows_per_sst = 800000 +row_group_size = 8192 +series_count = 100_000 +start_unix_nanos = 1_704_067_200_000_000_000 # 2024-01-01T00:00:00Z +step_nanos = 15_000_000_000 +time_range_layout = "non_overlapping_per_sst" +series_layout = "timestamp_major" + +[[scenario.queries]] +name = "selector_full" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069900, '15s') promql_constant_tag_concat_ms_100k" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "selector_host1" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069900, '15s') promql_constant_tag_concat_ms_100k{host='host1'}" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "topk_5_full" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069900, '15s') topk(5, promql_constant_tag_concat_ms_100k)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "topk_5_host1" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069900, '15s') topk(5, promql_constant_tag_concat_ms_100k{host='host1'})" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "topk_100_full" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069900, '15s') topk(100, promql_constant_tag_concat_ms_100k)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +[[scenario.queries]] +name = "bottomk_5_full" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069900, '15s') bottomk(5, promql_constant_tag_concat_ms_100k)" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# One evaluation time is a likely single-batch selector control, not a guarantee: +# PromQL lookback can still require multiple batches. +[[scenario.queries]] +name = "instant_selector_single_evaluation_control" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069000, '15s') promql_constant_tag_concat_ms_100k" +warmup = 3 +iterations = 9 + +[scenario.queries.thresholds] +max_candidate_latency_regression_pct = 10 + +# A single evaluation for output-parity inspection, separate from timed controls. +[[scenario.queries]] +name = "instant_topk_5_output_parity_control" +kind = "tql" +query = "TQL EVAL (1704069000, 1704069000, '15s') topk(5, promql_constant_tag_concat_ms_100k)" +warmup = 0 +iterations = 1 diff --git a/tests/perf/query_cases/promql_constant_tag_concat/case.toml b/tests/perf/query_cases/promql_constant_tag_concat_ms_10k/case.toml similarity index 81% rename from tests/perf/query_cases/promql_constant_tag_concat/case.toml rename to tests/perf/query_cases/promql_constant_tag_concat_ms_10k/case.toml index e28f771ead3..0046fa6742c 100644 --- a/tests/perf/query_cases/promql_constant_tag_concat/case.toml +++ b/tests/perf/query_cases/promql_constant_tag_concat_ms_10k/case.toml @@ -1,13 +1,13 @@ -# CI performance qualification for the constant-tag concat experiment. +# Millisecond timestamp high-cardinality qualification for constant-tag concat. # # This direct-SST fixture fixes file, row-group, and label layout independently # of ingestion and SQL flush behavior; it does not claim to reproduce a # SQL-flush baseline. With timestamp-major generation, every series occurs at -# every scrape timestamp and each series spans all 32 non-overlapping SSTs, -# exercising cross-file reads for the multi-evaluation queries. +# every scrape timestamp and each series has eight samples in all 32 +# non-overlapping SSTs, exercising cross-file reads for the multi-evaluation queries. [case] -name = "promql_constant_tag_concat" +name = "promql_constant_tag_concat_ms_10k" description = "PromQL selector and topk/bottomk qualification on a controlled constant-tag direct-SST fixture" [scenario] @@ -16,7 +16,7 @@ seed = 4096 [[scenario.tables]] database = "public" -name = "promql_constant_tag_concat" +name = "promql_constant_tag_concat_ms_10k" engine = "mito" append_mode = true sst_format = "flat" @@ -33,7 +33,7 @@ distribution = { kind = "cardinality", values = 16, prefix = "host" } name = "instance" type = "STRING" semantic = "tag" -distribution = { kind = "cardinality", values = 4096, prefix = "instance" } +distribution = { kind = "cardinality", values = 10_000, prefix = "instance" } [[scenario.tables.columns]] name = "reading" @@ -43,15 +43,15 @@ distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 } [[scenario.tables.columns]] name = "ts" -type = "TIMESTAMP(9)" +type = "TIMESTAMP(3)" semantic = "timestamp" [scenario.layout] regions = 1 sst_count = 32 -rows_per_sst = 32768 +rows_per_sst = 80000 row_group_size = 8192 -series_count = 4096 +series_count = 10_000 start_unix_nanos = 1_704_067_200_000_000_000 # 2024-01-01T00:00:00Z step_nanos = 15_000_000_000 time_range_layout = "non_overlapping_per_sst" @@ -60,7 +60,7 @@ series_layout = "timestamp_major" [[scenario.queries]] name = "selector_full" kind = "tql" -query = "TQL EVAL (1704069000, 1704069900, '15s') promql_constant_tag_concat" +query = "TQL EVAL (1704069000, 1704069900, '15s') promql_constant_tag_concat_ms_10k" warmup = 3 iterations = 9 @@ -70,7 +70,7 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "selector_host1" kind = "tql" -query = "TQL EVAL (1704069000, 1704069900, '15s') promql_constant_tag_concat{host='host1'}" +query = "TQL EVAL (1704069000, 1704069900, '15s') promql_constant_tag_concat_ms_10k{host='host1'}" warmup = 3 iterations = 9 @@ -80,7 +80,7 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "topk_5_full" kind = "tql" -query = "TQL EVAL (1704069000, 1704069900, '15s') topk(5, promql_constant_tag_concat)" +query = "TQL EVAL (1704069000, 1704069900, '15s') topk(5, promql_constant_tag_concat_ms_10k)" warmup = 3 iterations = 9 @@ -90,7 +90,7 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "topk_5_host1" kind = "tql" -query = "TQL EVAL (1704069000, 1704069900, '15s') topk(5, promql_constant_tag_concat{host='host1'})" +query = "TQL EVAL (1704069000, 1704069900, '15s') topk(5, promql_constant_tag_concat_ms_10k{host='host1'})" warmup = 3 iterations = 9 @@ -100,7 +100,7 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "topk_100_full" kind = "tql" -query = "TQL EVAL (1704069000, 1704069900, '15s') topk(100, promql_constant_tag_concat)" +query = "TQL EVAL (1704069000, 1704069900, '15s') topk(100, promql_constant_tag_concat_ms_10k)" warmup = 3 iterations = 9 @@ -110,7 +110,7 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "bottomk_5_full" kind = "tql" -query = "TQL EVAL (1704069000, 1704069900, '15s') bottomk(5, promql_constant_tag_concat)" +query = "TQL EVAL (1704069000, 1704069900, '15s') bottomk(5, promql_constant_tag_concat_ms_10k)" warmup = 3 iterations = 9 @@ -122,7 +122,7 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "instant_selector_single_evaluation_control" kind = "tql" -query = "TQL EVAL (1704069000, 1704069000, '15s') promql_constant_tag_concat" +query = "TQL EVAL (1704069000, 1704069000, '15s') promql_constant_tag_concat_ms_10k" warmup = 3 iterations = 9 @@ -133,6 +133,6 @@ max_candidate_latency_regression_pct = 10 [[scenario.queries]] name = "instant_topk_5_output_parity_control" kind = "tql" -query = "TQL EVAL (1704069000, 1704069000, '15s') topk(5, promql_constant_tag_concat)" +query = "TQL EVAL (1704069000, 1704069000, '15s') topk(5, promql_constant_tag_concat_ms_10k)" warmup = 0 iterations = 1