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

Merge branch 'work/9070-native-lastrow' into perf/pr9070-otlp-query

Browse Source
This commit is contained in:
discord9
2026-09-10 17:50:21 +08:00
parent 4b020fe5b3 10eda8d580
commit 478ba6507c
5 changed files with 592 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/promql/src/extension_plan/range_manipulate.rs
+75 -11
View File
@@ -761,16 +761,7 @@ impl RangeManipulateStream {
let start = query_start.max(first_ts_aligned);
let end = query_end.min(last_ts_aligned);
if start > end {
let bounds = if start >= i64::MIN as i128
&& start <= i64::MAX as i128
&& end >= i64::MIN as i128
&& end <= i64::MAX as i128
{
(start as i64, end as i64)
} else {
(self.start, self.end)
};
return Ok((vec![], bounds));
return Ok((vec![], (self.start, self.end)));
}
// The intersection is within the declared i64 query bounds.
let start = start as i64;
@@ -1508,6 +1499,68 @@ mod test {
}
}
#[tokio::test]
async fn no_intersection_batch_is_skipped_and_stream_continues() {
let schema = Arc::new(Schema::new(vec![
Field::new(
TIME_INDEX_COLUMN,
TimestampMillisecondType::DATA_TYPE,
false,
),
Field::new("value", DataType::Float64, false),
]));
let input = LogicalPlan::EmptyRelation(EmptyRelation {
produce_one_row: false,
schema: schema.clone().to_dfschema_ref().unwrap(),
});
let plan = RangeManipulate::new(
0,
50,
10,
1,
TIME_INDEX_COLUMN.to_string(),
vec!["value".to_string()],
input,
)
.unwrap();
let no_intersection = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondArray::from(vec![100])),
Arc::new(Float64Array::from(vec![1.0])),
],
)
.unwrap();
let intersection = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(TimestampMillisecondArray::from(vec![20])),
Arc::new(Float64Array::from(vec![2.0])),
],
)
.unwrap();
let input = Arc::new(DataSourceExec::new(Arc::new(
MemorySourceConfig::try_new(&[vec![no_intersection, intersection]], schema, None)
.unwrap(),
)));
let batches = datafusion::physical_plan::collect(
plan.to_execution_plan(input),
SessionContext::default().task_ctx(),
)
.await
.unwrap();
assert_eq!(batches.len(), 1);
assert_eq!(batches[0].num_rows(), 1);
let timestamps = batches[0]
.column(0)
.as_any()
.downcast_ref::<TimestampMillisecondArray>()
.unwrap();
assert_eq!(timestamps.value(0), 20);
}
fn calculate_range_for_test(
query_start: i64,
query_end: i64,
@@ -1634,7 +1687,7 @@ mod test {
10,
0,
vec![100],
(100, 50),
(0, 50),
),
];
@@ -1772,6 +1825,17 @@ mod test {
let (actual, (start, end)) =
calculate_range_for_test(query_start, query_end, interval, range, &timestamps);
let expected = calculate_range_oracle(&timestamps, start, end, interval, range);
let expected = if actual.is_empty() && !expected.is_empty() {
assert!(
expected.iter().all(|(_, len)| *len == 0),
"case={case}, timestamps={timestamps:?}, query=({query_start}, {query_end}), \
interval={interval}, range={range}, bounds=({start}, {end}): \
no-intersection output must have no selected samples"
);
vec![]
} else {
expected
};
assert_eq!(
actual, expected,
"case={case}, timestamps={timestamps:?}, query=({query_start}, {query_end}), \
src/query/src/optimizer/scan_hint.rs
+116 -39
View File
@@ -19,7 +19,6 @@ use arrow_schema::SortOptions;
use common_function::aggrs::aggr_wrapper::aggr_state_func_name;
use common_recordbatch::OrderOption;
use common_recordbatch::filter::SimpleFilterEvaluator;
use common_time::timestamp::TimeUnit;
use datafusion::datasource::DefaultTableSource;
use datafusion_common::tree_node::{Transformed, TreeNodeRewriter};
use datafusion_common::{Column, Result};
@@ -139,8 +138,7 @@ impl ScanHintRule {
/// predicate later rejects that row. Only recognized tag/time predicates are
/// allowed: tags select whole series, and supported time predicates constrain
/// the scan window before row selection. Field or unrecognized predicates are
/// conservatively rejected. Finer-than-millisecond timestamps are also excluded
/// because instant evaluation can conflate distinct samples at that precision.
/// conservatively rejected.
///
/// This checks only attached predicates; the path allowlist separately rejects
/// residual Filter nodes between InstantManipulate and the scan.
@@ -149,14 +147,6 @@ impl ScanHintRule {
provider: &DummyTableProvider,
) -> bool {
let metadata = provider.region_metadata();
// Instant evaluation is millisecond-based, so finer time units can
// conflate timestamps and must not use the LastRow hint.
if !matches!(
metadata.time_index_type().unit(),
TimeUnit::Second | TimeUnit::Millisecond
) {
return false;
}
for filter in &table_scan.filters {
let Some(filter) = SimpleFilterEvaluator::try_new(filter) else {
return false;
@@ -682,26 +672,49 @@ mod test {
}
fn last_value_aggregate(input: LogicalPlan) -> LogicalPlan {
LogicalPlanBuilder::from(input)
let aggregate = LogicalPlanBuilder::from(input)
.aggregate(
vec![col("k0")],
vec![Expr::AggregateFunction(AggregateFunction {
func: last_value_udaf(),
params: AggregateFunctionParams {
args: vec![col("v0")],
distinct: false,
filter: None,
order_by: vec![Sort {
expr: col("ts"),
asc: true,
nulls_first: true,
}],
null_treatment: None,
},
})],
vec![
Expr::AggregateFunction(AggregateFunction {
func: last_value_udaf(),
params: AggregateFunctionParams {
args: vec![col("v0")],
distinct: false,
filter: None,
order_by: vec![Sort {
expr: col("ts"),
asc: true,
nulls_first: true,
}],
null_treatment: None,
},
}),
Expr::AggregateFunction(AggregateFunction {
func: last_value_udaf(),
params: AggregateFunctionParams {
args: vec![col("ts")],
distinct: false,
filter: None,
order_by: vec![Sort {
expr: col("ts"),
asc: true,
nulls_first: true,
}],
null_treatment: None,
},
}),
],
)
.unwrap()
.build()
.unwrap();
let timestamp = aggregate.schema().field(2).name().clone();
LogicalPlanBuilder::from(aggregate)
.project(vec![col("k0"), col(timestamp).alias("ts")])
.unwrap()
.build()
.unwrap()
}
@@ -928,21 +941,39 @@ mod test {
None,
)
.unwrap()
.project(vec![
Expr::Column(Column::new(Some("left"), "ts")),
Expr::Column(Column::new(Some("left"), "v0")),
])
.unwrap()
.build()
.unwrap();
let nonlast_aggregate = LogicalPlanBuilder::from(scan_plan(provider(), "aggregate"))
.aggregate(
vec![col("k0")],
vec![Expr::AggregateFunction(AggregateFunction {
func: max_udaf(),
params: AggregateFunctionParams {
args: vec![col("v0")],
distinct: false,
filter: None,
order_by: vec![],
null_treatment: None,
},
})],
vec![
Expr::AggregateFunction(AggregateFunction {
func: max_udaf(),
params: AggregateFunctionParams {
args: vec![col("v0")],
distinct: false,
filter: None,
order_by: vec![],
null_treatment: None,
},
}),
Expr::AggregateFunction(AggregateFunction {
func: max_udaf(),
params: AggregateFunctionParams {
args: vec![col("ts")],
distinct: false,
filter: None,
order_by: vec![],
null_treatment: None,
},
})
.alias("ts"),
],
)
.unwrap()
.build()
@@ -1023,7 +1054,49 @@ mod test {
}
#[test]
fn single_evaluation_rejects_microsecond_and_nanosecond_time_index_casts() {
fn single_evaluation_uses_last_row_for_microsecond_and_nanosecond_time_indexes() {
for timestamp_type in [
ConcreteDataType::timestamp_microsecond_datatype(),
ConcreteDataType::timestamp_nanosecond_datatype(),
] {
let direct_provider = Arc::new(mock_table_provider_with_timestamp(
RegionId::new(1, 1),
timestamp_type.clone(),
));
let direct = ScanHintRule
.rewrite(
single_evaluation(scan_plan(direct_provider, "direct")),
&OptimizerContext::default(),
)
.unwrap()
.data;
assert_eq!(
scan_requests(&direct)[0].series_row_selector,
Some(TimeSeriesRowSelector::LastRow { after_merge: true })
);
let projection_provider = Arc::new(mock_table_provider_with_timestamp(
RegionId::new(1, 1),
timestamp_type,
));
let projection = LogicalPlanBuilder::from(scan_plan(projection_provider, "projection"))
.project(vec![col("ts")])
.unwrap()
.build()
.unwrap();
let projected = ScanHintRule
.rewrite(single_evaluation(projection), &OptimizerContext::default())
.unwrap()
.data;
assert_eq!(
scan_requests(&projected)[0].series_row_selector,
Some(TimeSeriesRowSelector::LastRow { after_merge: true })
);
}
}
#[test]
fn single_evaluation_rejects_lossy_microsecond_and_nanosecond_time_index_casts() {
for timestamp_type in [
ConcreteDataType::timestamp_microsecond_datatype(),
ConcreteDataType::timestamp_nanosecond_datatype(),
@@ -1081,7 +1154,7 @@ mod test {
#[test]
fn single_evaluation_rejects_projection_expressions_that_change_rows() {
let invalid_projections = [
vec![col("ts").alias("renamed")],
vec![col("ts").alias("renamed"), col("ts")],
vec![
Expr::BinaryExpr(datafusion_expr::expr::BinaryExpr::new(
Box::new(col("v0")),
@@ -1089,6 +1162,7 @@ mod test {
Box::new(lit(1.0_f64)),
))
.alias("v0"),
col("ts"),
],
vec![
Expr::Cast(Cast::new(
@@ -1097,7 +1171,10 @@ mod test {
))
.alias("ts"),
],
vec![Expr::Cast(Cast::new(Box::new(col("v0")), DataType::Int64)).alias("v0")],
vec![
Expr::Cast(Cast::new(Box::new(col("v0")), DataType::Int64)).alias("v0"),
col("ts"),
],
vec![
Expr::Cast(Cast::new(
Box::new(col("ts")),
src/query/src/promql/planner.rs
+96 -39
View File
@@ -2786,13 +2786,14 @@ impl PromPlanner {
_ => None,
})
.unwrap_or(ArrowTimeUnit::Millisecond);
let scalar = |milliseconds: i64| -> Option<ScalarValue> {
let value = match unit {
ArrowTimeUnit::Second => milliseconds.div_euclid(1_000),
ArrowTimeUnit::Millisecond => milliseconds,
ArrowTimeUnit::Microsecond => milliseconds.checked_mul(1_000)?,
ArrowTimeUnit::Nanosecond => milliseconds.checked_mul(1_000_000)?,
};
let native_value = |milliseconds: i128| match unit {
ArrowTimeUnit::Second => milliseconds.div_euclid(1_000),
ArrowTimeUnit::Millisecond => milliseconds,
ArrowTimeUnit::Microsecond => milliseconds * 1_000,
ArrowTimeUnit::Nanosecond => milliseconds * 1_000_000,
};
let scalar = |milliseconds: i128| -> Option<ScalarValue> {
let value = i64::try_from(native_value(milliseconds)).ok()?;
Some(match unit {
ArrowTimeUnit::Second => ScalarValue::TimestampSecond(Some(value), None),
ArrowTimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(value), None),
@@ -2801,45 +2802,57 @@ impl PromPlanner {
})
};
let window = self.ctx.range.unwrap_or(self.ctx.lookback_delta);
let filter = |lower_ms: i64, upper_ms: i64| -> Option<DfExpr> {
let lower = DfExpr::Literal(scalar(lower_ms)?, None);
let lower_filter = if window == 0 {
time_index_expr.clone().gt_eq(lower)
} else if unit == ArrowTimeUnit::Millisecond
&& let Some(inclusive_lower) = lower_ms.checked_add(1)
{
time_index_expr
.clone()
.gt_eq(DfExpr::Literal(scalar(inclusive_lower)?, None))
} else {
time_index_expr.clone().gt(lower)
};
Some(
lower_filter.and(
let filter = |lower_ms: i128, upper_ms: i128| {
let lower_value = native_value(lower_ms);
let upper_value = native_value(upper_ms);
if lower_value > i128::from(i64::MAX) || upper_value < i128::from(i64::MIN) {
return Some(lit(false));
}
let lower_filter = (lower_value >= i128::from(i64::MIN)).then(|| {
let lower = DfExpr::Literal(scalar(lower_ms).unwrap(), None);
if window == 0 {
time_index_expr.clone().gt_eq(lower)
} else if unit == ArrowTimeUnit::Millisecond
&& let Some(inclusive_lower) = lower_ms
.checked_add(1)
.and_then(|lower| i64::try_from(lower).ok())
.and_then(|lower| scalar(i128::from(lower)))
{
time_index_expr
.clone()
.lt_eq(DfExpr::Literal(scalar(upper_ms)?, None)),
),
)
.gt_eq(DfExpr::Literal(inclusive_lower, None))
} else {
time_index_expr.clone().gt(lower)
}
});
let upper_filter = (upper_value <= i128::from(i64::MAX)).then(|| {
time_index_expr
.clone()
.lt_eq(DfExpr::Literal(scalar(upper_ms).unwrap(), None))
});
match (lower_filter, upper_filter) {
(Some(lower), Some(upper)) => Some(lower.and(upper)),
(Some(filter), None) | (None, Some(filter)) => Some(filter),
(None, None) => None,
}
};
let bounds = |timestamp: i64| {
timestamp
.checked_sub(offset_duration)
.and_then(|upper| upper.checked_sub(window).map(|lower| (lower, upper)))
let upper = i128::from(timestamp) - i128::from(offset_duration);
(upper - i128::from(window), upper)
};
let num_points = (end as i128 - start as i128) / self.ctx.interval as i128;
if num_points > MAX_SCATTER_POINTS as i128 || self.ctx.interval <= INTERVAL_1H {
return Ok(bounds(start)
.zip(bounds(end))
.and_then(|((lower, _), (_, upper))| filter(lower, upper)));
let (lower, _) = bounds(start);
let (_, upper) = bounds(end);
return Ok(filter(lower, upper));
}
let mut filters = Vec::new();
for timestamp in (start..=end).step_by(self.ctx.interval as usize) {
let Some((lower, upper)) = bounds(timestamp) else {
// An unrepresentable envelope must not discard samples.
return Ok(None);
};
let (lower, upper) = bounds(timestamp);
let Some(filter) = filter(lower, upper) else {
// A point whose native bounds cannot be represented may cover the whole native
// time domain, so its disjunct cannot be omitted.
return Ok(None);
};
filters.push(filter);
@@ -12254,11 +12267,55 @@ mod test {
);
}
planner.ctx.end = i64::MAX;
let filter = planner
.build_time_index_filter(0, &schema)
.unwrap()
.unwrap()
.to_string();
assert!(
planner
.build_time_index_filter(0, &schema)
.unwrap()
.is_none()
filter.contains("timestamp >= TimestampNanosecond(1000000000, None)"),
"{filter}"
);
// A lookback subtraction can underflow milliseconds while the upper bound remains
// representable. Keep that upper bound so LastRow cannot select a future sample.
let ms_schema = Arc::new(
DFSchema::try_from(ArrowSchema::new(vec![Field::new(
"timestamp",
ArrowDataType::Timestamp(ArrowTimeUnit::Millisecond, None),
false,
)]))
.unwrap(),
);
planner.ctx.start = i64::MIN + 100;
planner.ctx.end = planner.ctx.start;
planner.ctx.lookback_delta = 200;
let filter = planner
.build_time_index_filter(0, &ms_schema)
.unwrap()
.unwrap()
.to_string();
assert_eq!(
filter,
format!(
"timestamp <= TimestampMillisecond({}, None)",
i64::MIN + 100
)
);
// The lower bound can also overflow while converting milliseconds to native nanoseconds.
// Its representable upper bound still has to reach the scan.
planner.ctx.start = 0;
planner.ctx.end = 0;
planner.ctx.lookback_delta = 300_000;
let filter = planner
.build_time_index_filter(9_223_372_036_854, &schema)
.unwrap()
.unwrap()
.to_string();
assert_eq!(
filter,
"timestamp <= TimestampNanosecond(-9223372036854000000, None)"
);
}
tests/cases/standalone/common/promql/native_time_selection.result
+220 -4
View File
@@ -30,6 +30,67 @@ INSERT INTO native_time_us VALUES
Affected Rows: 18
-- The native projection and exact 1ms-lookback bounds must reach the scan;
-- the 1s+tick row must not displace 201.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
-- SQLNESS REPLACE native_time_us.__table_id\s*=\s*UInt32\(\d+\) native_time_us.__table_id=UInt32(REDACTED)
TQL EXPLAIN (1, 1, '1s', '1ms') native_time_us{series="exact"};
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | MergeScan [is_placeholder=false, remote_input=[ |
| | PromInstantManipulate: range=[1000..1000], lookback=[1], interval=[1000], time index=[ts] |
| | PromSeriesDivide: tags=["series"] |
| | Sort: native_time_us.series ASC NULLS FIRST, native_time_us.ts ASC NULLS FIRST |
| | Projection: native_time_us.val, native_time_us.series, native_time_us.ts |
| | Filter: native_time_us.series = Utf8("exact") AND native_time_us.ts > TimestampMicrosecond(999000, None) AND native_time_us.ts <= TimestampMicrosecond(1000000, None) |
| | TableScan: native_time_us, partial_filters=[native_time_us.series = Utf8("exact"), native_time_us.ts > TimestampMicrosecond(999000, None), native_time_us.ts <= TimestampMicrosecond(1000000, None)] |
| | ]] |
| physical_plan | CooperativeExec |
| | MergeScanExec: REDACTED
| | |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
-- The actual memtable scan must use LastRow { after_merge: true } with native
-- 1ms-lookback bounds; it must select exact 1s rather than the future tick.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (flat_format.*) REDACTED
-- SQLNESS REPLACE (elapsed_compute.*) REDACTED
TQL ANALYZE VERBOSE (1, 1, '1s', '1ms') native_time_us{series="exact"};
+-+-+-+
| stage | node | plan_|
+-+-+-+
| 0_| 0_|_CooperativeExec metrics=[]_|
|_|_|_MergeScanExec: REDACTED
|_|_|_|
| 1_| 0_|_PromInstantManipulateExec: range=[1000..1000], lookback=[1], interval=[1000], time index=[ts] metrics=[output_rows: 1, REDACTED
|_|_|_PromSeriesDivideExec: tags=["series"] metrics=[output_rows: 1, REDACTED
|_|_|_ProjectionExec: expr=[val@2 as val, series@1 as series, ts@0 as ts] metrics=[output_rows: 1, REDACTED
|_|_|_CooperativeExec metrics=[]_|
|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "series", "val"], "filters": ["series = Dictionary(UInt32, Utf8(\"exact\"))", "ts > TimestampMicrosecond(999000, None)", "ts <= TimestampMicrosecond(1000000, None)"], "REDACTED
|_|_|_|
|_|_| Total rows: 1_|
+-+-+-+
-- The same-series future tick is in the memtable, while exact 1s remains selected.
TQL EVAL (1, 1, '1s', '1ms') native_time_us{series="exact"};
+-------+--------+---------------------+
| val | series | ts |
+-------+--------+---------------------+
| 201.0 | exact | 1970-01-01T00:00:01 |
+-------+--------+---------------------+
-- Future-only selection is empty before flushing, exercising the memtable path.
TQL EVAL (1, 1, '1s', '300s') native_time_us{series="future"};
@@ -44,8 +105,8 @@ ADMIN FLUSH_TABLE('native_time_us');
| 0 |
+-------------------------------------+
-- At 1s, selection keeps an exact native timestamp.
TQL EVAL (1, 1, '1s', '300s') native_time_us{series="exact"};
-- The exact native sample remains selected from the flushed SST.
TQL EVAL (1, 1, '1s', '1ms') native_time_us{series="exact"};
+-------+--------+---------------------+
| val | series | ts |
@@ -195,6 +256,67 @@ INSERT INTO native_time_ns VALUES
Affected Rows: 18
-- The native projection and exact 1ms-lookback bounds must reach the scan;
-- the 1s+tick row must not displace 201.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
-- SQLNESS REPLACE native_time_ns.__table_id\s*=\s*UInt32\(\d+\) native_time_ns.__table_id=UInt32(REDACTED)
TQL EXPLAIN (1, 1, '1s', '1ms') native_time_ns{series="exact"};
+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | MergeScan [is_placeholder=false, remote_input=[ |
| | PromInstantManipulate: range=[1000..1000], lookback=[1], interval=[1000], time index=[ts] |
| | PromSeriesDivide: tags=["series"] |
| | Sort: native_time_ns.series ASC NULLS FIRST, native_time_ns.ts ASC NULLS FIRST |
| | Projection: native_time_ns.val, native_time_ns.series, native_time_ns.ts |
| | Filter: native_time_ns.series = Utf8("exact") AND native_time_ns.ts > TimestampNanosecond(999000000, None) AND native_time_ns.ts <= TimestampNanosecond(1000000000, None) |
| | TableScan: native_time_ns, partial_filters=[native_time_ns.series = Utf8("exact"), native_time_ns.ts > TimestampNanosecond(999000000, None), native_time_ns.ts <= TimestampNanosecond(1000000000, None)] |
| | ]] |
| physical_plan | CooperativeExec |
| | MergeScanExec: REDACTED
| | |
+---------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
-- The actual memtable scan must use LastRow { after_merge: true } with native
-- 1ms-lookback bounds; it must select exact 1s rather than the future tick.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (flat_format.*) REDACTED
-- SQLNESS REPLACE (elapsed_compute.*) REDACTED
TQL ANALYZE VERBOSE (1, 1, '1s', '1ms') native_time_ns{series="exact"};
+-+-+-+
| stage | node | plan_|
+-+-+-+
| 0_| 0_|_CooperativeExec metrics=[]_|
|_|_|_MergeScanExec: REDACTED
|_|_|_|
| 1_| 0_|_PromInstantManipulateExec: range=[1000..1000], lookback=[1], interval=[1000], time index=[ts] metrics=[output_rows: 1, REDACTED
|_|_|_PromSeriesDivideExec: tags=["series"] metrics=[output_rows: 1, REDACTED
|_|_|_ProjectionExec: expr=[val@2 as val, series@1 as series, ts@0 as ts] metrics=[output_rows: 1, REDACTED
|_|_|_CooperativeExec metrics=[]_|
|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "series", "val"], "filters": ["series = Dictionary(UInt32, Utf8(\"exact\"))", "ts > TimestampNanosecond(999000000, None)", "ts <= TimestampNanosecond(1000000000, None)"], "REDACTED
|_|_|_|
|_|_| Total rows: 1_|
+-+-+-+
-- The same-series future tick is in the memtable, while exact 1s remains selected.
TQL EVAL (1, 1, '1s', '1ms') native_time_ns{series="exact"};
+-------+--------+---------------------+
| val | series | ts |
+-------+--------+---------------------+
| 201.0 | exact | 1970-01-01T00:00:01 |
+-------+--------+---------------------+
-- Future-only selection is empty before flushing, exercising the memtable path.
TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="future"};
@@ -209,8 +331,8 @@ ADMIN FLUSH_TABLE('native_time_ns');
| 0 |
+-------------------------------------+
-- At 1s, selection keeps an exact native timestamp.
TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="exact"};
-- The exact native sample remains selected from the flushed SST.
TQL EVAL (1, 1, '1s', '1ms') native_time_ns{series="exact"};
+-------+--------+---------------------+
| val | series | ts |
@@ -330,6 +452,100 @@ DROP TABLE native_time_ns;
Affected Rows: 0
-- An unrepresentable native lower bound must not discard its representable upper bound.
-- The upper filter must reach LastRow so the 1ms-future row cannot hide the eligible row.
CREATE TABLE native_time_ns_lower_overflow (
ts TIMESTAMP(9) TIME INDEX,
series STRING PRIMARY KEY,
val DOUBLE,
);
Affected Rows: 0
INSERT INTO native_time_ns_lower_overflow VALUES
(-9223200000000000000, 'exact', 1),
(-9223199999999000000, 'exact', 2);
Affected Rows: 2
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE native_time_ns_lower_overflow.__table_id\s*=\s*UInt32\(\d+\) native_time_ns_lower_overflow.__table_id=UInt32(REDACTED)
TQL EXPLAIN (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | MergeScan [is_placeholder=false, remote_input=[ |
| | PromInstantManipulate: range=[0..0], lookback=[172800000], interval=[1000], time index=[ts] |
| | PromSeriesNormalize: offset=[9223200000000], time index=[ts], filter NaN: [false] |
| | PromSeriesDivide: tags=["series"] |
| | Sort: native_time_ns_lower_overflow.series ASC NULLS FIRST, native_time_ns_lower_overflow.ts ASC NULLS FIRST |
| | Projection: native_time_ns_lower_overflow.val, native_time_ns_lower_overflow.series, native_time_ns_lower_overflow.ts |
| | Filter: native_time_ns_lower_overflow.series = Utf8("exact") AND native_time_ns_lower_overflow.ts <= TimestampNanosecond(-9223200000000000000, None) |
| | TableScan: native_time_ns_lower_overflow, partial_filters=[native_time_ns_lower_overflow.series = Utf8("exact"), native_time_ns_lower_overflow.ts <= TimestampNanosecond(-9223200000000000000, None)] |
| | ]] |
| physical_plan | CooperativeExec |
| | MergeScanExec: REDACTED
| | |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (flat_format.*) REDACTED
-- SQLNESS REPLACE (elapsed_compute.*) REDACTED
TQL ANALYZE VERBOSE (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
+-+-+-+
| stage | node | plan_|
+-+-+-+
| 0_| 0_|_CooperativeExec metrics=[]_|
|_|_|_MergeScanExec: REDACTED
|_|_|_|
| 1_| 0_|_PromInstantManipulateExec: range=[0..0], lookback=[172800000], interval=[1000], time index=[ts] metrics=[output_rows: 1, REDACTED
|_|_|_PromSeriesNormalizeExec: offset=[9223200000000], time index=[ts], filter NaN: [false] metrics=[output_rows: 1, REDACTED
|_|_|_PromSeriesDivideExec: tags=["series"] metrics=[output_rows: 1, REDACTED
|_|_|_ProjectionExec: expr=[val@2 as val, series@1 as series, ts@0 as ts] metrics=[output_rows: 1, REDACTED
|_|_|_CooperativeExec metrics=[]_|
|_|_|_SeriesScan: region=REDACTED, {"partition_count":{"count":1, "mem_ranges":1, "files":0, "file_ranges":0}, "selector":"LastRow { after_merge: true }", "distribution":"PerSeries", "projection": ["ts", "series", "val"], "filters": ["series = Dictionary(UInt32, Utf8(\"exact\"))", "ts <= TimestampNanosecond(-9223200000000000000, None)"], "REDACTED
|_|_|_|
|_|_| Total rows: 1_|
+-+-+-+
-- The representable upper bound selects only the exact row.
TQL EVAL (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
+-----+--------+---------------------+
| val | series | ts |
+-----+--------+---------------------+
| 1.0 | exact | 1970-01-01T00:00:00 |
+-----+--------+---------------------+
ADMIN FLUSH_TABLE('native_time_ns_lower_overflow');
+----------------------------------------------------+
| ADMIN FLUSH_TABLE('native_time_ns_lower_overflow') |
+----------------------------------------------------+
| 0 |
+----------------------------------------------------+
TQL EVAL (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
+-----+--------+---------------------+
| val | series | ts |
+-----+--------+---------------------+
| 1.0 | exact | 1970-01-01T00:00:00 |
+-----+--------+---------------------+
DROP TABLE native_time_ns_lower_overflow;
Affected Rows: 0
-- Second precision is promoted before applying fractional-second offsets.
CREATE TABLE native_time_sec (ts TIMESTAMP(0) TIME INDEX, val DOUBLE);
tests/cases/standalone/common/promql/native_time_selection.sql
+85 -4
View File
@@ -27,13 +27,37 @@ INSERT INTO native_time_us VALUES
(1000000, 'window', 3),
(1000001, 'window', 4);
-- The native projection and exact 1ms-lookback bounds must reach the scan;
-- the 1s+tick row must not displace 201.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
-- SQLNESS REPLACE native_time_us.__table_id\s*=\s*UInt32\(\d+\) native_time_us.__table_id=UInt32(REDACTED)
TQL EXPLAIN (1, 1, '1s', '1ms') native_time_us{series="exact"};
-- The actual memtable scan must use LastRow { after_merge: true } with native
-- 1ms-lookback bounds; it must select exact 1s rather than the future tick.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (flat_format.*) REDACTED
-- SQLNESS REPLACE (elapsed_compute.*) REDACTED
TQL ANALYZE VERBOSE (1, 1, '1s', '1ms') native_time_us{series="exact"};
-- The same-series future tick is in the memtable, while exact 1s remains selected.
TQL EVAL (1, 1, '1s', '1ms') native_time_us{series="exact"};
-- Future-only selection is empty before flushing, exercising the memtable path.
TQL EVAL (1, 1, '1s', '300s') native_time_us{series="future"};
ADMIN FLUSH_TABLE('native_time_us');
-- At 1s, selection keeps an exact native timestamp.
TQL EVAL (1, 1, '1s', '300s') native_time_us{series="exact"};
-- The exact native sample remains selected from the flushed SST.
TQL EVAL (1, 1, '1s', '1ms') native_time_us{series="exact"};
TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_us{series="future"});
TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_us{series="exact"});
@@ -89,13 +113,37 @@ INSERT INTO native_time_ns VALUES
(1000000000, 'window', 3),
(1000000001, 'window', 4);
-- The native projection and exact 1ms-lookback bounds must reach the scan;
-- the 1s+tick row must not displace 201.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (RepartitionExec:.*) RepartitionExec: REDACTED
-- SQLNESS REPLACE native_time_ns.__table_id\s*=\s*UInt32\(\d+\) native_time_ns.__table_id=UInt32(REDACTED)
TQL EXPLAIN (1, 1, '1s', '1ms') native_time_ns{series="exact"};
-- The actual memtable scan must use LastRow { after_merge: true } with native
-- 1ms-lookback bounds; it must select exact 1s rather than the future tick.
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (flat_format.*) REDACTED
-- SQLNESS REPLACE (elapsed_compute.*) REDACTED
TQL ANALYZE VERBOSE (1, 1, '1s', '1ms') native_time_ns{series="exact"};
-- The same-series future tick is in the memtable, while exact 1s remains selected.
TQL EVAL (1, 1, '1s', '1ms') native_time_ns{series="exact"};
-- Future-only selection is empty before flushing, exercising the memtable path.
TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="future"};
ADMIN FLUSH_TABLE('native_time_ns');
-- At 1s, selection keeps an exact native timestamp.
TQL EVAL (1, 1, '1s', '300s') native_time_ns{series="exact"};
-- The exact native sample remains selected from the flushed SST.
TQL EVAL (1, 1, '1s', '1ms') native_time_ns{series="exact"};
TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_ns{series="future"});
TQL EVAL (1, 1, '1s', '300s') timestamp(native_time_ns{series="exact"});
@@ -125,6 +173,39 @@ TQL EVAL (1, 1, '1s') last_over_time((native_time_ns{series="exact"})[1s:1s]);
DROP TABLE native_time_ns;
-- An unrepresentable native lower bound must not discard its representable upper bound.
-- The upper filter must reach LastRow so the 1ms-future row cannot hide the eligible row.
CREATE TABLE native_time_ns_lower_overflow (
ts TIMESTAMP(9) TIME INDEX,
series STRING PRIMARY KEY,
val DOUBLE,
);
INSERT INTO native_time_ns_lower_overflow VALUES
(-9223200000000000000, 'exact', 1),
(-9223199999999000000, 'exact', 2);
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE native_time_ns_lower_overflow.__table_id\s*=\s*UInt32\(\d+\) native_time_ns_lower_overflow.__table_id=UInt32(REDACTED)
TQL EXPLAIN (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
-- SQLNESS REPLACE (RoundRobinBatch.*) REDACTED
-- SQLNESS REPLACE (Hash.*) REDACTED
-- SQLNESS REPLACE (-+) -
-- SQLNESS REPLACE (\s\s+) _
-- SQLNESS REPLACE (peers.*) REDACTED
-- SQLNESS REPLACE region=\d+\(\d+,\s+\d+\) region=REDACTED
-- SQLNESS REPLACE (flat_format.*) REDACTED
-- SQLNESS REPLACE (elapsed_compute.*) REDACTED
TQL ANALYZE VERBOSE (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
-- The representable upper bound selects only the exact row.
TQL EVAL (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
ADMIN FLUSH_TABLE('native_time_ns_lower_overflow');
TQL EVAL (0, 0, '1s', '2d') native_time_ns_lower_overflow{series="exact"} offset 106750d;
DROP TABLE native_time_ns_lower_overflow;
-- Second precision is promoted before applying fractional-second offsets.
CREATE TABLE native_time_sec (ts TIMESTAMP(0) TIME INDEX, val DOUBLE);
INSERT INTO native_time_sec VALUES (0, 10), (1, 11), (2, 12);