mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-13 07:22:32 +00:00
fix(promql): apply offset to subquery evaluation window (#9364)
* fix(promql): apply offset to subquery evaluation window `prom_subquery_expr_to_plan` destructured `SubqueryExpr` without reading `offset`, so `<subquery>[range:step] offset <d>` planned exactly the same window as the un-offset form and silently returned data for the wrong time range. The plain vector/matrix-selector paths already threaded the offset through `selector_to_series_normalize_plan` and `RangeManipulate`. Shift the inner evaluation window back by the offset and pass the offset to the subquery's `RangeManipulate`, which maps the inner samples forward onto the evaluation timeline before bucketing them into ranges. This matches Prometheus, whose `evaluator.subqueryTimeRange` evaluates the inner expression over `(start - offset - range, end - offset]` and whose `evalSubquery` then hands the samples to the outer range-vector function as a `MatrixSelector` that still carries the subquery offset. An offset on the inner selector composes additively, as `subqueryTimes` documents. `RangeManipulate`'s protobuf message has no offset field and recovers it on decode from an immediately underlying `SeriesNormalize`. Since `RangeManipulate` is commutative in `dist_plan` and can be pushed below a `MergeScan`, insert that carrier node so the offset survives distributed planning instead of decoding as zero. Known divergence, unchanged by this commit: Prometheus anchors subquery step points on absolute epoch multiples of the step, while GreptimeDB anchors them on the evaluation start. The two agree whenever the offset is a multiple of the subquery step; the added sqlness cases stay within that range. Closes #9330 Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Aarav <aaravsjadav@gmail.com> * docs(promql): correct the subquery-offset rationale and pin the histogram shift Follow-up toe132c267, which stated two things inaccurately and left one path untested. No behaviour change beyond a conditional clone. The `SeriesNormalize` carrier was described as existing "so the offset survives distributed planning", implying standalone is unaffected. That is wrong. `local_offset` is read on the substrait decode path (`range_manipulate.rs`, `instant_manipulate.rs`), and GreptimeDB routes PromQL plans through `MergeScan`/substrait in standalone too -- the standalone results `promql/encode_substrait.result` and `precisions.result` both show `MergeScan [is_placeholder=false, remote_input=[...]]`. Deleting the carrier therefore empties subquery-with-offset results in standalone as well, while `count_over_time_subquery_with_offset` keeps passing because it asserts the pre-serialization plan. Say so at the node, so the next reader does not remove it believing standalone is safe. The carrier is also not semantically inert: `SeriesNormalizeStream::normalize` shifts native histogram `start_timestamp` payloads by `offset`. On the subquery path that runs on a computed inner result, on top of any offset the inner selector already applied, and nothing pinned it. It is consistent -- `RangeManipulate` moves the millisecond time index forward by the same `offset_ms` -- so the distance from a histogram's start timestamp to the sample carrying it, which reset/rate detection reads, is invariant. Add `subquery_offset_shifts_histogram_start_and_time_index_together`, which executes the subquery's node stack over native histogram samples for a zero and a non-zero inner offset. It cannot be a sqlness case: native histograms are a struct column with no SQL type or literal (`sql_data_type_to_concrete_data_type` rejects structs) and the sqlness runner speaks only MySQL/Postgres, so such rows only arrive over gRPC/remote-write v2. The divergence note blamed sub-step offsets. Measured, that condition is too narrow: an unaligned evaluation timestamp alone diverges, with no offset at all -- `sum_over_time(fine[20s:10s])` at t=57 samples 47s and 57s here against Prometheus's 40s and 50s. The real condition is that `(start - offset)` is not a multiple of the subquery step, and it is pre-existing: the committed `tql eval (359, 359, '1s') sum_over_time(metric_total[60s:10s])` case is already an instance of it (`359 - 60 + 10 = 309`). Reword the comment and the sqlness header accordingly. A sub-step offset stays accepted: `foo[20s:10s] offset 5s` is valid PromQL, erroring on it would be a regression and would not close the gap anyway. This repo has no PromQL-compatibility documentation page, so there is nowhere else to record it. Also stop cloning the series key columns when there is no offset, and stop the `offset 0s` sqlness comment implying its parser error is evidence about this fix -- the same message comes from a `u32` overflow in `promql-parser`'s shared duration check (`offset 9999999999d`). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Signed-off-by: Aarav <aaravsjadav@gmail.com> * refactor(promql): trim subquery-offset comments and tests per review Shorten the planner and sqlness comments to the semantic contract, build `SeriesDivide` with a clone instead of an `Option`, and drop the normalize.rs histogram test and the `offset 0s` parser case. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bvab9gpXYjKyLN8Mo6JWWL Signed-off-by: Aarav <aaravsjadav@gmail.com> * fix(promql): record the subquery offset as the range fold offset The subquery's `RangeManipulate` now shifts the payload timestamps by the offset, so `predict_linear` over an offset subquery must recover the evaluation time with that offset instead of 0. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bvab9gpXYjKyLN8Mo6JWWL Signed-off-by: Aarav <aaravsjadav@gmail.com> --------- Signed-off-by: Aarav <aaravsjadav@gmail.com> Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com> (cherry picked from commitd6474b961c) Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
DeviousCardi
WenyXu
@@ -403,18 +403,34 @@ impl PromPlanner {
|
||||
subquery_expr: &SubqueryExpr,
|
||||
) -> Result<LogicalPlan> {
|
||||
let SubqueryExpr {
|
||||
expr, range, step, ..
|
||||
expr,
|
||||
range,
|
||||
step,
|
||||
offset,
|
||||
..
|
||||
} = subquery_expr;
|
||||
|
||||
// Prometheus evaluates the inner expression over `(start - offset - range, end - offset]`
|
||||
// (`subqueryTimeRange`). Shift the inner window back here; `RangeManipulate` maps the
|
||||
// samples forward again by the same offset.
|
||||
let offset_ms = match offset {
|
||||
Some(Offset::Pos(duration)) => duration.as_millis() as Millisecond,
|
||||
Some(Offset::Neg(duration)) => -(duration.as_millis() as Millisecond),
|
||||
None => 0,
|
||||
};
|
||||
|
||||
let current_interval = self.ctx.interval;
|
||||
if let Some(step) = step {
|
||||
self.ctx.interval = step.as_millis() as _;
|
||||
}
|
||||
let current_start = self.ctx.start;
|
||||
self.ctx.start -= range.as_millis() as i64 - self.ctx.interval;
|
||||
let current_end = self.ctx.end;
|
||||
self.ctx.start -= offset_ms + range.as_millis() as i64 - self.ctx.interval;
|
||||
self.ctx.end -= offset_ms;
|
||||
let input = self.prom_expr_to_plan(expr, query_engine_state).await?;
|
||||
self.ctx.interval = current_interval;
|
||||
self.ctx.start = current_start;
|
||||
self.ctx.end = current_end;
|
||||
|
||||
ensure!(!range.is_zero(), ZeroRangeSelectorSnafu);
|
||||
let range_ms = range.as_millis() as _;
|
||||
@@ -471,26 +487,42 @@ impl PromPlanner {
|
||||
.context(DataFusionPlanningSnafu)?;
|
||||
let divide_plan = LogicalPlan::Extension(Extension {
|
||||
node: Arc::new(SeriesDivide::new(
|
||||
series_key_columns,
|
||||
series_key_columns.clone(),
|
||||
time_index_column.clone(),
|
||||
sort_plan,
|
||||
)),
|
||||
});
|
||||
|
||||
// `RangeManipulate` has no offset in its protobuf message; decoding recovers it from the
|
||||
// `SeriesNormalize` directly below. Stale markers are not filtered: the input is computed.
|
||||
let divide_plan = if offset_ms != 0 {
|
||||
LogicalPlan::Extension(Extension {
|
||||
node: Arc::new(SeriesNormalize::new(
|
||||
offset_ms,
|
||||
time_index_column.clone(),
|
||||
false,
|
||||
series_key_columns,
|
||||
divide_plan,
|
||||
)),
|
||||
})
|
||||
} else {
|
||||
divide_plan
|
||||
};
|
||||
|
||||
let manipulate = RangeManipulate::new(
|
||||
self.ctx.start,
|
||||
self.ctx.end,
|
||||
self.ctx.interval,
|
||||
0,
|
||||
offset_ms,
|
||||
range_ms,
|
||||
time_index_column,
|
||||
self.ctx.field_columns.clone(),
|
||||
divide_plan,
|
||||
)
|
||||
.context(DataFusionPlanningSnafu)?;
|
||||
// A subquery always folds with offset 0, so its payload timestamps are already on the
|
||||
// evaluation timeline a function above it reads; see [`Self::create_range_eval_ts_expr`].
|
||||
self.ctx.range_fold_offset = Some(0);
|
||||
// The payload timestamps are shifted by the subquery offset; see
|
||||
// [`Self::create_range_eval_ts_expr`].
|
||||
self.ctx.range_fold_offset = Some(offset_ms);
|
||||
|
||||
Ok(LogicalPlan::Extension(Extension {
|
||||
node: Arc::new(manipulate),
|
||||
|
||||
@@ -5109,6 +5109,28 @@ async fn count_over_time_subquery() {
|
||||
indie_query_plan_compare(query, expected).await;
|
||||
}
|
||||
|
||||
/// `offset` on a subquery must shift the inner evaluation window back and be
|
||||
/// carried into the outer range manipulation. See
|
||||
/// <https://github.com/GreptimeTeam/greptimedb/issues/9330>.
|
||||
#[tokio::test]
|
||||
async fn count_over_time_subquery_with_offset() {
|
||||
let query = "count_over_time(some_metric[10m:1m] offset 5m)";
|
||||
let expected = String::from(
|
||||
"Filter: prom_count_over_time(timestamp_range,field_0) IS NOT NULL [timestamp:Timestamp(ms), prom_count_over_time(timestamp_range,field_0):Float64;N, tag_0:Utf8]\
|
||||
\n Projection: some_metric.timestamp, prom_count_over_time(timestamp_range, field_0) AS prom_count_over_time(timestamp_range,field_0), some_metric.tag_0 [timestamp:Timestamp(ms), prom_count_over_time(timestamp_range,field_0):Float64;N, tag_0:Utf8]\
|
||||
\n PromRangeManipulate: req range=[0..100000000], interval=[5000], eval range=[600000], time index=[timestamp], values=[\"field_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Dictionary(Int64, Float64);N, timestamp_range:Dictionary(Int64, Timestamp(ms))]\
|
||||
\n PromSeriesNormalize: offset=[300000], time index=[timestamp], filter NaN: [false] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n PromSeriesDivide: tags=[\"tag_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n PromInstantManipulate: range=[-840000..99700000], lookback=[1000], interval=[60000], time index=[timestamp] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n PromSeriesDivide: tags=[\"tag_0\"] [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n Sort: some_metric.tag_0 ASC NULLS FIRST, some_metric.timestamp ASC NULLS FIRST [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n Filter: some_metric.timestamp >= TimestampMillisecond(-840999, None) AND some_metric.timestamp <= TimestampMillisecond(99700000, None) [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]\
|
||||
\n TableScan: some_metric [tag_0:Utf8, timestamp:Timestamp(ms), field_0:Float64;N]",
|
||||
);
|
||||
indie_query_plan_compare(query, expected).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hash_join() {
|
||||
let mut eval_stmt = EvalStmt {
|
||||
|
||||
@@ -63,3 +63,103 @@ drop table metric_total;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
-- Offset on a subquery shifts the subquery's own evaluation window back by the offset.
|
||||
-- Reference: Prometheus `evaluator.subqueryTimeRange` (promql/engine.go).
|
||||
-- The offset cases stay on the subquery step grid so they match Prometheus directly.
|
||||
create table subquery_offset_total (
|
||||
ts timestamp time index,
|
||||
host string primary key,
|
||||
val double,
|
||||
);
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
insert into subquery_offset_total values
|
||||
(0, 'a', 1),
|
||||
(10000, 'a', 2),
|
||||
(20000, 'a', 3),
|
||||
(30000, 'a', 4),
|
||||
(40000, 'a', 5),
|
||||
(50000, 'a', 6),
|
||||
(60000, 'a', 7);
|
||||
|
||||
Affected Rows: 7
|
||||
|
||||
-- baseline: no offset at t=60 covers the 10s subquery points in (40s, 60s] -> 6 + 7
|
||||
tql eval (60, 60, '1s') sum_over_time(subquery_offset_total[20s:10s]);
|
||||
|
||||
+---------------------+----------------------------------+------+
|
||||
| ts | prom_sum_over_time(ts_range,val) | host |
|
||||
+---------------------+----------------------------------+------+
|
||||
| 1970-01-01T00:01:00 | 13.0 | a |
|
||||
+---------------------+----------------------------------+------+
|
||||
|
||||
-- the same subquery evaluated at t=30 -> 3 + 4
|
||||
tql eval (30, 30, '1s') sum_over_time(subquery_offset_total[20s:10s]);
|
||||
|
||||
+---------------------+----------------------------------+------+
|
||||
| ts | prom_sum_over_time(ts_range,val) | host |
|
||||
+---------------------+----------------------------------+------+
|
||||
| 1970-01-01T00:00:30 | 7.0 | a |
|
||||
+---------------------+----------------------------------+------+
|
||||
|
||||
-- `offset 30s` at t=60 must equal the un-offset subquery at t=30
|
||||
tql eval (60, 60, '1s') sum_over_time(subquery_offset_total[20s:10s] offset 30s);
|
||||
|
||||
+---------------------+----------------------------------+------+
|
||||
| ts | prom_sum_over_time(ts_range,val) | host |
|
||||
+---------------------+----------------------------------+------+
|
||||
| 1970-01-01T00:01:00 | 7.0 | a |
|
||||
+---------------------+----------------------------------+------+
|
||||
|
||||
-- a negative offset looks ahead of the evaluation time
|
||||
tql eval (30, 30, '1s') sum_over_time(subquery_offset_total[20s:10s] offset -30s);
|
||||
|
||||
+---------------------+----------------------------------+------+
|
||||
| ts | prom_sum_over_time(ts_range,val) | host |
|
||||
+---------------------+----------------------------------+------+
|
||||
| 1970-01-01T00:00:30 | 13.0 | a |
|
||||
+---------------------+----------------------------------+------+
|
||||
|
||||
-- an offset on the inner selector composes additively with the subquery offset: Prometheus
|
||||
-- `subqueryTimes` accumulates "the sum of offsets and ranges of all subqueries in the path",
|
||||
-- and the inner selector subtracts its own offset from the already shifted step timestamps.
|
||||
-- 20s + 10s therefore behaves like the un-offset subquery at t=30.
|
||||
tql eval (60, 60, '1s') sum_over_time((subquery_offset_total offset 10s)[20s:10s] offset 20s);
|
||||
|
||||
+---------------------+----------------------------------+------+
|
||||
| ts | prom_sum_over_time(ts_range,val) | host |
|
||||
+---------------------+----------------------------------+------+
|
||||
| 1970-01-01T00:01:00 | 7.0 | a |
|
||||
+---------------------+----------------------------------+------+
|
||||
|
||||
-- ... and the inner offset alone accounts for the same total shift
|
||||
tql eval (60, 60, '1s') sum_over_time((subquery_offset_total offset 30s)[20s:10s]);
|
||||
|
||||
+---------------------+----------------------------------+------+
|
||||
| ts | prom_sum_over_time(ts_range,val) | host |
|
||||
+---------------------+----------------------------------+------+
|
||||
| 1970-01-01T00:01:00 | 7.0 | a |
|
||||
+---------------------+----------------------------------+------+
|
||||
|
||||
-- `predict_linear` predicts from the evaluation time: 4 + 0.1 * 30 = 7 and 7 - 0.1 * 30 = 4
|
||||
tql eval (60, 60, '1s') predict_linear(subquery_offset_total[20s:10s] offset 30s, 0);
|
||||
|
||||
+---------------------+----------------------------------------------+------+
|
||||
| ts | prom_predict_linear(ts_range,val,Float64(0)) | host |
|
||||
+---------------------+----------------------------------------------+------+
|
||||
| 1970-01-01T00:01:00 | 7.0 | a |
|
||||
+---------------------+----------------------------------------------+------+
|
||||
|
||||
tql eval (30, 30, '1s') predict_linear(subquery_offset_total[20s:10s] offset -30s, 0);
|
||||
|
||||
+---------------------+----------------------------------------------+------+
|
||||
| ts | prom_predict_linear(ts_range,val,Float64(0)) | host |
|
||||
+---------------------+----------------------------------------------+------+
|
||||
| 1970-01-01T00:00:30 | 4.0 | a |
|
||||
+---------------------+----------------------------------------------+------+
|
||||
|
||||
drop table subquery_offset_total;
|
||||
|
||||
Affected Rows: 0
|
||||
|
||||
|
||||
@@ -20,3 +20,49 @@ tql eval (10, 10, '1s') rate(metric_total[20s:10s]);
|
||||
tql eval (20, 20, '1s') rate(metric_total[20s:5s]);
|
||||
|
||||
drop table metric_total;
|
||||
|
||||
-- Offset on a subquery shifts the subquery's own evaluation window back by the offset.
|
||||
-- Reference: Prometheus `evaluator.subqueryTimeRange` (promql/engine.go).
|
||||
-- The offset cases stay on the subquery step grid so they match Prometheus directly.
|
||||
create table subquery_offset_total (
|
||||
ts timestamp time index,
|
||||
host string primary key,
|
||||
val double,
|
||||
);
|
||||
|
||||
insert into subquery_offset_total values
|
||||
(0, 'a', 1),
|
||||
(10000, 'a', 2),
|
||||
(20000, 'a', 3),
|
||||
(30000, 'a', 4),
|
||||
(40000, 'a', 5),
|
||||
(50000, 'a', 6),
|
||||
(60000, 'a', 7);
|
||||
|
||||
-- baseline: no offset at t=60 covers the 10s subquery points in (40s, 60s] -> 6 + 7
|
||||
tql eval (60, 60, '1s') sum_over_time(subquery_offset_total[20s:10s]);
|
||||
|
||||
-- the same subquery evaluated at t=30 -> 3 + 4
|
||||
tql eval (30, 30, '1s') sum_over_time(subquery_offset_total[20s:10s]);
|
||||
|
||||
-- `offset 30s` at t=60 must equal the un-offset subquery at t=30
|
||||
tql eval (60, 60, '1s') sum_over_time(subquery_offset_total[20s:10s] offset 30s);
|
||||
|
||||
-- a negative offset looks ahead of the evaluation time
|
||||
tql eval (30, 30, '1s') sum_over_time(subquery_offset_total[20s:10s] offset -30s);
|
||||
|
||||
-- an offset on the inner selector composes additively with the subquery offset: Prometheus
|
||||
-- `subqueryTimes` accumulates "the sum of offsets and ranges of all subqueries in the path",
|
||||
-- and the inner selector subtracts its own offset from the already shifted step timestamps.
|
||||
-- 20s + 10s therefore behaves like the un-offset subquery at t=30.
|
||||
tql eval (60, 60, '1s') sum_over_time((subquery_offset_total offset 10s)[20s:10s] offset 20s);
|
||||
|
||||
-- ... and the inner offset alone accounts for the same total shift
|
||||
tql eval (60, 60, '1s') sum_over_time((subquery_offset_total offset 30s)[20s:10s]);
|
||||
|
||||
-- `predict_linear` predicts from the evaluation time: 4 + 0.1 * 30 = 7 and 7 - 0.1 * 30 = 4
|
||||
tql eval (60, 60, '1s') predict_linear(subquery_offset_total[20s:10s] offset 30s, 0);
|
||||
|
||||
tql eval (30, 30, '1s') predict_linear(subquery_offset_total[20s:10s] offset -30s, 0);
|
||||
|
||||
drop table subquery_offset_total;
|
||||
|
||||
Reference in New Issue
Block a user