rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-12 06:52:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(promql): address constant-tag review and cardinality coverage

Browse Source 
Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com>
This commit is contained in:
discord9
2026-09-15 12:17:12 +08:00
parent fcf12bfc76
commit e12fd660fc
4 changed files with 205 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/extension_plan/series_divide.rs
+50 -1
View File
@@ -521,13 +521,32 @@ fn concat_series_batches(
batches: &[RecordBatch],
tag_indices: &[usize],
) -> DataFusionResult<RecordBatch> {
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 [
tests/perf/query_cases/promql_constant_tag_concat_ms/case.toml
-1
View File
@@ -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
tests/perf/query_cases/promql_constant_tag_concat_ms_100k/case.toml
+138
View File
@@ -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
tests/perf/query_cases/promql_constant_tag_concat/case.toml → tests/perf/query_cases/promql_constant_tag_concat_ms_10k/case.toml
+17 -17
View File
@@ -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