mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-12 06:52:34 +00:00
perf(promql): reduce timestamp payload allocations and use narrow cursor scans (#9417)
* perf(promql): zero-copy timestamp payload for millisecond zero-offset ranges RangeManipulate rebuilt the timestamp range payload per sample through wide-precision i128 conversion even for millisecond input with zero offset, where the input is already millisecond and nothing changes. That turned a previously zero-copy path into a per-sample loop plus a Vec allocation, regressing range-function queries. Reuse the input timestamp array directly when the time index is already millisecond and the offset is zero. Other units and non-zero offsets keep the wide-precision conversion (required so ns/us data near the native type limits stays correct under large offsets). Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 <discord9@163.com> * perf(promql): vectorize timestamp payload with plain i64 RangeManipulate rebuilt the timestamp range payload per sample through wide-precision i128 conversion, turning a hot path into a scalar loop with a Vec allocation. Realistic timestamps and offsets stay far inside i64, so the wide path is unnecessary. - millisecond + zero offset reuses the input array with zero copy; - other units / offsets shift in one vectorized i64 pass (no i128, no per-sample Option, no overflow branch). Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 <discord9@163.com> * perf(promql): check timestamp payload overflow via arrow try_unary Replace the silent wrapping shift with try_unary + checked_mul/checked_add. checked arithmetic lowers to a per-lane overflow flag so the single pass still vectorizes over the dense i64 buffer; any overflow short-circuits to an error instead of silently wrapping to a wrong timestamp. No i128, no per-sample Option. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 <discord9@163.com> * perf(promql): vectorize cursor scans with i64 fast path RangeManipulate::calculate_range and InstantManipulate::manipulate compared every sample through i128 wide-precision arithmetic. For realistic timestamps and offsets the shifted values stay far inside i64, so the wide path was pure overhead. Split both scans into an i64 fast path (checked arithmetic precomputed once per batch) and an i128 exact fallback. Any input that would overflow i64 falls back to the i128 scan, keeping window membership semantics identical. Verified: 47/47 range_manipulate + instant_manipulate tests, 275/275 promql crate tests, 1612 randomized cases byte-identical to i128 scan. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 <discord9@163.com> * refactor(promql): generalize cursor scans and share endpoint precheck Address review comments on PR #9417: - Generalize scan_ranges/scan_instants over Nanoseconds (i64/i128), eliminating the duplicated scan bodies. - Extract shared narrow_scan precheck used by both RangeManipulate and InstantManipulate, with identical checked endpoint logic. - Trim doc comments to one line; fix factual errors about fallback trigger (2262 for ms/s, not native type limits). Semantics unchanged: same cursor algorithm, same overflow fallback, same i64-vs-i128 selection, same stale/null/tie handling. Verified: 47/47 range_manipulate + instant_manipulate tests, clippy clean. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 <discord9@163.com> * perf(promql): vectorize payload with try_unary and add scan carrier bench Replace the per-sample i128 payload loop with a single arrow try_unary pass: i128 intermediate holds any native timestamp shift, only the final millisecond result must fit i64. This is semantically equivalent to the original i128 implementation (no silent wrap, no per-sample branching). Add cursor_scan_carriers A/B benchmark: i64 fast path vs i128 exact fallback on four workloads. Add five deterministic differential tests pinning scan_ranges::<i64> and scan_ranges::<i128> to be pointwise equivalent on boundary shapes (sub-ms samples, fallback triggers, nanosecond extremes, seeded precision matrix). Bench (N=4096, 15s cadence): - payload wall_time 20x15s: 74.94us -> 48.98us (-35%) - payload wall_time 240x15s: 102.44us - scan i64 fast path: 10.97us vs i128 19.84us (1.81x) Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * perf(promql): restore zero-copy payload for ms+offset=0 and fix comment Restore the zero-copy fast path for millisecond timestamps with zero offset, guarded by the input column's exact data type (Timestamp (Millisecond, None)) rather than self.time_unit. The old time_unit guard would accept a tz-carrying millisecond column and produce a dictionary value type disagreeing with the output schema; the datatype guard is the correct check. Also fix the comment: arrow's try_unary is a fallible kernel that LLVM cannot auto-vectorize; the win over the hand-written loop comes from eliminating the per-sample Option/Vec and branches, not SIMD. Add a regression test with a timezone-carrying millisecond index to prove it falls back to the try_unary path. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * perf(promql): replace try_unary with unary + Cell overflow flag Arrow's try_unary is a fallible kernel that LLVM cannot auto-vectorize. Replace it with the infallible unary kernel: the closure is a plain i64 -> i64 mapping that sets a Cell<bool> on overflow instead of returning Err per sample. The flag is checked once after the loop. This removes the per-lane Result construction and lets LLVM vectorize the shift+truncate sequence. Bench (N=4096, 15s cadence): - window4x15s: 42.12us -> 28.77us (-32%) - window20x15s: 48.98us -> 27.68us (-43%) - window240x15s: 102.44us -> 27.61us (-73%) - regular_sparse: 90.04us -> 29.20us (-68%) All cases now run at ~27-29us regardless of window size; the zero-copy path dominates for ms+offset=0. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * perf(promql): dispatch payload conversion on null_count; drop wall-time bench Null-free batches (dominant PromQL case) take arrow's vectorizable unary kernel with a Cell<bool> overflow flag; nullable batches take try_unary so a NULL slot's arbitrary underlying value cannot raise a false batch-wide overflow error. Add a regression test exercising both paths. Remove bench_range_manipulate_wall_time (all-ms zero-offset cases only exercised the zero-copy path, measuring nothing about the conversion) and its orphaned helpers; restore the cursor-scan bench helpers it shared. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> * fix(promql): keep nullable timestamp cursor scans on i128 NULL slots may contain unordered arbitrary values, so the endpoint-only narrow-scan proof cannot bound their interior timestamp arithmetic. Both range and instant scans now attempt i64 only for null-free timestamp arrays and otherwise retain the existing exact i128 path. Add focused interior-NULL regressions through both active entrypoints, checking concrete selections and output against the existing wide scan. The null-free fast path and timestamp payload conversion are unchanged. Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> --------- Signed-off-by: discord9 <55937128+discord9@users.noreply.github.com> Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
discord9
@@ -17,25 +17,14 @@
|
||||
use std::sync::Arc;
|
||||
|
||||
use criterion::{BenchmarkId, Criterion, criterion_group};
|
||||
use datafusion::arrow::array::{
|
||||
Array, ArrayRef, DictionaryArray, Float64Array, TimestampMillisecondArray,
|
||||
};
|
||||
use datafusion::arrow::datatypes::{
|
||||
DataType as ArrowDataType, Field as ArrowField, Int64Type, Schema,
|
||||
};
|
||||
use datafusion::arrow::record_batch::RecordBatch;
|
||||
use datafusion::common::ToDFSchema;
|
||||
use datafusion::datasource::memory::MemorySourceConfig;
|
||||
use datafusion::datasource::source::DataSourceExec;
|
||||
use datafusion::execution::context::TaskContext;
|
||||
use datafusion::logical_expr::{EmptyRelation, LogicalPlan};
|
||||
use datafusion::physical_plan::{ColumnarValue, ExecutionPlan};
|
||||
use datafusion::arrow::array::{Array, Float64Array, TimestampMillisecondArray};
|
||||
use datafusion::arrow::datatypes::TimeUnit;
|
||||
use datafusion::physical_plan::ColumnarValue;
|
||||
use datafusion_common::ScalarValue;
|
||||
use datafusion_common::config::ConfigOptions;
|
||||
use datafusion_expr::ScalarFunctionArgs;
|
||||
use datatypes::arrow::datatypes::{DataType, Field};
|
||||
use futures::StreamExt;
|
||||
use promql::extension_plan::RangeManipulate;
|
||||
use promql::extension_plan::RangeManipulateStream;
|
||||
use promql::functions::{
|
||||
AbsentOverTime, Changes, CountOverTime, Delta, DoubleExponentialSmoothing, IDelta, Increase,
|
||||
LastOverTime, MaxOverTime, MinOverTime, PredictLinear, PresentOverTime, QuantileOverTime, Rate,
|
||||
@@ -964,223 +953,294 @@ fn bench_edge_count_functions(c: &mut Criterion) {
|
||||
group.finish();
|
||||
}
|
||||
|
||||
const RANGE_MANIPULATE_CADENCE_MS: i64 = 15_000;
|
||||
/// One cursor-scan workload: the timeline, the window and the aligned instants to visit.
|
||||
#[derive(Clone, Debug)]
|
||||
struct CursorScanWorkload {
|
||||
time_unit: TimeUnit,
|
||||
offset_ms: i64,
|
||||
timestamps: Vec<i64>,
|
||||
window_ms: i64,
|
||||
interval_ms: i64,
|
||||
start_ms: i64,
|
||||
end_ms: i64,
|
||||
}
|
||||
|
||||
fn make_range_manipulate_batch(timestamps: Vec<i64>, field_count: usize) -> RecordBatch {
|
||||
let mut fields = Vec::with_capacity(field_count + 1);
|
||||
let mut columns: Vec<ArrayRef> = Vec::with_capacity(field_count + 1);
|
||||
fields.push(ArrowField::new(
|
||||
"timestamp",
|
||||
ArrowDataType::Timestamp(datafusion::arrow::datatypes::TimeUnit::Millisecond, None),
|
||||
false,
|
||||
));
|
||||
columns.push(Arc::new(TimestampMillisecondArray::from(timestamps)) as _);
|
||||
|
||||
for field_index in 0..field_count {
|
||||
fields.push(ArrowField::new(
|
||||
format!("value_{field_index}"),
|
||||
ArrowDataType::Float64,
|
||||
false,
|
||||
));
|
||||
let values = (0..columns[0].len())
|
||||
.map(|row| row as f64 + field_index as f64 * 0.01)
|
||||
.collect::<Vec<_>>();
|
||||
columns.push(Arc::new(Float64Array::from(values)) as _);
|
||||
impl CursorScanWorkload {
|
||||
/// The workload through the narrow `i64` carrier, or `None` when production would fall
|
||||
/// back to `i128` for it.
|
||||
///
|
||||
/// For the fallback cases the fast path is timed on a twin workload: the same shape
|
||||
/// translated onto a timeline the narrow carrier holds, which keeps the number of sample
|
||||
/// and window comparisons identical to the fallback run.
|
||||
fn bench_scan_narrow(&self) -> Option<Vec<(u32, u32)>> {
|
||||
RangeManipulateStream::bench_scan_narrow(
|
||||
self.time_unit,
|
||||
self.offset_ms,
|
||||
self.window_ms,
|
||||
self.interval_ms,
|
||||
&self.timestamps,
|
||||
self.start_ms,
|
||||
self.end_ms,
|
||||
)
|
||||
}
|
||||
|
||||
RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap()
|
||||
}
|
||||
/// The workload through the exact `i128` carrier, the fallback production takes
|
||||
/// whenever [`Self::bench_scan_narrow`] returns `None`.
|
||||
fn bench_scan_wide(&self) -> Vec<(u32, u32)> {
|
||||
RangeManipulateStream::bench_scan_wide(
|
||||
self.time_unit,
|
||||
self.offset_ms,
|
||||
self.window_ms,
|
||||
self.interval_ms,
|
||||
&self.timestamps,
|
||||
self.start_ms,
|
||||
self.end_ms,
|
||||
)
|
||||
}
|
||||
|
||||
async fn execute_and_drain(
|
||||
plan: Arc<dyn ExecutionPlan>,
|
||||
task_ctx: Arc<TaskContext>,
|
||||
) -> Vec<RecordBatch> {
|
||||
let mut stream = plan.execute(0, task_ctx).unwrap();
|
||||
let mut output = Vec::new();
|
||||
while let Some(batch) = stream.next().await {
|
||||
output.push(batch.unwrap());
|
||||
fn evaluations(&self) -> usize {
|
||||
((self.end_ms - self.start_ms) / self.interval_ms) as usize + 1
|
||||
}
|
||||
|
||||
fn label(&self) -> String {
|
||||
format!(
|
||||
"N{}_eval{}_window{}x{}ms",
|
||||
self.timestamps.len(),
|
||||
self.evaluations(),
|
||||
self.window_ms / self.interval_ms,
|
||||
self.interval_ms
|
||||
)
|
||||
}
|
||||
output
|
||||
}
|
||||
|
||||
fn range_keys(batch: &RecordBatch, index: usize) -> Vec<(u32, u32)> {
|
||||
let ranges = batch
|
||||
.column(index)
|
||||
.as_any()
|
||||
.downcast_ref::<DictionaryArray<Int64Type>>()
|
||||
.unwrap()
|
||||
.clone();
|
||||
RangeArray::try_new(ranges)
|
||||
.unwrap()
|
||||
.ranges()
|
||||
.map(Option::unwrap)
|
||||
.collect()
|
||||
/// Cadence of the cursor-scan benchmarks: PromQL's usual 15s step.
|
||||
const CURSOR_SCAN_CADENCE_MS: i64 = 15_000;
|
||||
|
||||
/// Milliseconds of a one-hour window, the widest window benchmarked here.
|
||||
const CURSOR_SCAN_HOUR_MS: i64 = 3_600_000;
|
||||
|
||||
/// Builds a workload from a first timestamp and a sample count.
|
||||
///
|
||||
/// `sample_spacing_native` is the distance between samples in native ticks: 15s worth, so
|
||||
/// every native-precision workload has the same shape (one sample per cadence, 20 samples
|
||||
/// per 300s window). The aligned instants start at the first shifted sample, so every
|
||||
/// window overlaps the input the same way in both arms of the A/B comparison.
|
||||
fn cursor_scan_workload(
|
||||
time_unit: TimeUnit,
|
||||
offset_ms: i64,
|
||||
first_timestamp: i64,
|
||||
samples: usize,
|
||||
window_ms: i64,
|
||||
sample_spacing_native: i64,
|
||||
) -> CursorScanWorkload {
|
||||
let interval_ms = CURSOR_SCAN_CADENCE_MS;
|
||||
let timestamps = (0..samples as i64)
|
||||
.map(|index| first_timestamp + index * sample_spacing_native)
|
||||
.collect::<Vec<_>>();
|
||||
// `Millisecond` instants of the first shifted sample, flooring like `calculate_range`.
|
||||
let start_ms = shifted_floor_millis(time_unit, offset_ms, first_timestamp);
|
||||
let evaluations = samples - 1;
|
||||
CursorScanWorkload {
|
||||
time_unit,
|
||||
offset_ms,
|
||||
timestamps,
|
||||
window_ms,
|
||||
interval_ms,
|
||||
start_ms,
|
||||
end_ms: start_ms + evaluations as i64 * interval_ms,
|
||||
}
|
||||
}
|
||||
|
||||
fn brute_force_range_keys(
|
||||
timestamps: &[i64],
|
||||
evaluations: usize,
|
||||
window_points: usize,
|
||||
) -> Vec<(u32, u32)> {
|
||||
let range = window_points as i64 * RANGE_MANIPULATE_CADENCE_MS;
|
||||
(0..evaluations)
|
||||
.map(|evaluation| {
|
||||
let current = evaluation as i64 * RANGE_MANIPULATE_CADENCE_MS;
|
||||
let mut offset = None;
|
||||
let mut length = 0;
|
||||
for (index, timestamp) in timestamps.iter().enumerate() {
|
||||
if *timestamp > current - range && *timestamp <= current {
|
||||
offset.get_or_insert(index as u32);
|
||||
length += 1;
|
||||
}
|
||||
}
|
||||
(offset.unwrap_or(0), length)
|
||||
})
|
||||
.collect()
|
||||
/// Native ticks of one benchmark cadence, i.e. 15s in the given precision.
|
||||
fn cadence_in_native_ticks(time_unit: TimeUnit) -> i64 {
|
||||
match time_unit {
|
||||
TimeUnit::Second => CURSOR_SCAN_CADENCE_MS / 1_000,
|
||||
TimeUnit::Millisecond => CURSOR_SCAN_CADENCE_MS,
|
||||
TimeUnit::Microsecond => CURSOR_SCAN_CADENCE_MS * 1_000,
|
||||
TimeUnit::Nanosecond => CURSOR_SCAN_CADENCE_MS * 1_000_000,
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_range_manipulate_output(
|
||||
output: &[RecordBatch],
|
||||
field_count: usize,
|
||||
expected_range_keys: Option<&[(u32, u32)]>,
|
||||
) {
|
||||
let Some(expected_range_keys) = expected_range_keys else {
|
||||
assert!(output.is_empty());
|
||||
return;
|
||||
/// Floor of a shifted sample in milliseconds, in exact `i128` native precision.
|
||||
fn shifted_floor_millis(time_unit: TimeUnit, offset_ms: i64, timestamp: i64) -> i64 {
|
||||
let scale = match time_unit {
|
||||
TimeUnit::Second => 1_000_000_000i128,
|
||||
TimeUnit::Millisecond => 1_000_000i128,
|
||||
TimeUnit::Microsecond => 1_000i128,
|
||||
TimeUnit::Nanosecond => 1i128,
|
||||
};
|
||||
i64::try_from(
|
||||
((timestamp as i128) * scale + (offset_ms as i128) * 1_000_000).div_euclid(1_000_000),
|
||||
)
|
||||
.expect("the shifted millisecond must stay legal")
|
||||
}
|
||||
|
||||
/// A/B benchmark of the cursor-scan carriers behind `RangeManipulate::scan_ranges`.
|
||||
///
|
||||
/// `RangeManipulate` scans the aligned instants in `i64` nanoseconds whenever
|
||||
/// `narrow_scan` proves the batch and the bounds fit, and in exact `i128` nanoseconds
|
||||
/// otherwise. Both arms below drive the production scan through the carrier the operator
|
||||
/// would pick, so the numbers isolate the carrier and not the surrounding operator work:
|
||||
///
|
||||
/// - `ms_zero_offset`: millisecond batch with no offset, the `i64` fast path. The same
|
||||
/// workload is also timed through `i128` to price the fallback on this input.
|
||||
/// - `ms_zero_offset_long_window`: same, with a one-hour window, where the scan advances
|
||||
/// both cursors much further per evaluation.
|
||||
/// - `ms_large_offset`: an offset whose nanosecond product overflows `i64`, so production
|
||||
/// falls back to `i128`. The `i64` arm times the identical workload shape translated onto
|
||||
/// a narrow timeline, which keeps the selections and thus the work identical.
|
||||
/// - `ns_native_floor`: nanosecond batch at the `i64::MIN` floor, where the shifted start
|
||||
/// bound underflows `i64` nanoseconds, so production falls back to `i128`. The `i64` arm
|
||||
/// again times the translated narrow shape.
|
||||
///
|
||||
/// Every arm asserts that both carriers selected the same ranges, so the comparison cannot
|
||||
/// be won by doing less work. The fallback triggers themselves are pinned by the unit tests
|
||||
/// in `range_manipulate.rs` (`scan_falls_back_to_i128_when_only_the_sample_tail_overflows`
|
||||
/// and `scan_carriers_split_at_the_nanosecond_boundaries`).
|
||||
fn bench_cursor_scan_carriers(c: &mut Criterion) {
|
||||
const SAMPLES: usize = 4_096;
|
||||
const SHORT_WINDOW_MS: i64 = 20 * CURSOR_SCAN_CADENCE_MS;
|
||||
// `i64::MAX` milliseconds is ~9.22e18 ms; one microsecond past it in the nanosecond
|
||||
// product makes `narrow_scan` reject the offset, while the shifted millisecond stays legal.
|
||||
const LARGE_OFFSET_MS: i64 = 9_223_372_036_855;
|
||||
const LARGE_OFFSET_BASE_MS: i64 = 1_000_000_000_000_000;
|
||||
const NANOSECOND_FLOOR_SHIFT_NS: i64 = 9_223_372_036_000_000_000;
|
||||
const NANOSECOND_FLOOR_SAMPLES: usize = 4_096;
|
||||
|
||||
let ms_zero_offset = cursor_scan_workload(
|
||||
TimeUnit::Millisecond,
|
||||
0,
|
||||
0,
|
||||
SAMPLES,
|
||||
SHORT_WINDOW_MS,
|
||||
cadence_in_native_ticks(TimeUnit::Millisecond),
|
||||
);
|
||||
let ms_zero_offset_long_window = cursor_scan_workload(
|
||||
TimeUnit::Millisecond,
|
||||
0,
|
||||
0,
|
||||
SAMPLES,
|
||||
CURSOR_SCAN_HOUR_MS,
|
||||
cadence_in_native_ticks(TimeUnit::Millisecond),
|
||||
);
|
||||
// The large offset is shifted inside one millisecond of the nanosecond ceiling, so the
|
||||
// payload stays legal: the fallback is driven by the product, not by an illegal instant.
|
||||
let ms_large_offset = cursor_scan_workload(
|
||||
TimeUnit::Millisecond,
|
||||
LARGE_OFFSET_MS,
|
||||
LARGE_OFFSET_BASE_MS,
|
||||
SAMPLES,
|
||||
SHORT_WINDOW_MS,
|
||||
cadence_in_native_ticks(TimeUnit::Millisecond),
|
||||
);
|
||||
// The narrow twin of `ms_large_offset`: the same batch shape without the offset.
|
||||
let ms_large_offset_narrow = cursor_scan_workload(
|
||||
TimeUnit::Millisecond,
|
||||
0,
|
||||
0,
|
||||
SAMPLES,
|
||||
SHORT_WINDOW_MS,
|
||||
cadence_in_native_ticks(TimeUnit::Millisecond),
|
||||
);
|
||||
|
||||
// Nanosecond batch on the native floor: samples at `i64::MIN + i * 15s`, whose shifted
|
||||
// start bound underflows `i64` nanoseconds.
|
||||
let nanosecond_floor_first = i64::MIN;
|
||||
let nanosecond_floor = cursor_scan_workload(
|
||||
TimeUnit::Nanosecond,
|
||||
0,
|
||||
nanosecond_floor_first,
|
||||
NANOSECOND_FLOOR_SAMPLES,
|
||||
SHORT_WINDOW_MS,
|
||||
cadence_in_native_ticks(TimeUnit::Nanosecond),
|
||||
);
|
||||
// The narrow twin: the same shape translated by a multiple of one millisecond, so both
|
||||
// the samples and the aligned instants keep their relative geometry.
|
||||
let nanosecond_floor_narrow = CursorScanWorkload {
|
||||
time_unit: TimeUnit::Nanosecond,
|
||||
offset_ms: 0,
|
||||
timestamps: nanosecond_floor
|
||||
.timestamps
|
||||
.iter()
|
||||
.map(|timestamp| timestamp + NANOSECOND_FLOOR_SHIFT_NS)
|
||||
.collect(),
|
||||
window_ms: nanosecond_floor.window_ms,
|
||||
interval_ms: nanosecond_floor.interval_ms,
|
||||
start_ms: nanosecond_floor.start_ms + NANOSECOND_FLOOR_SHIFT_NS / 1_000_000,
|
||||
end_ms: nanosecond_floor.end_ms + NANOSECOND_FLOOR_SHIFT_NS / 1_000_000,
|
||||
};
|
||||
|
||||
assert_eq!(output.len(), 1);
|
||||
let batch = &output[0];
|
||||
assert_eq!(batch.num_rows(), expected_range_keys.len());
|
||||
|
||||
let timestamp_range_keys = range_keys(batch, field_count + 1);
|
||||
assert_eq!(timestamp_range_keys, expected_range_keys);
|
||||
|
||||
for field_index in 0..field_count {
|
||||
assert_eq!(range_keys(batch, field_index + 1), expected_range_keys);
|
||||
}
|
||||
}
|
||||
|
||||
fn bench_range_manipulate_wall_time(c: &mut Criterion) {
|
||||
let mut group = c.benchmark_group("range_manipulate_wall_time");
|
||||
let primary_points = 4_096;
|
||||
let sparse_evaluations = primary_points - 1;
|
||||
let mut cases = vec![
|
||||
let cases = [
|
||||
(
|
||||
"one_field",
|
||||
(0..primary_points)
|
||||
.map(|point| point as i64 * RANGE_MANIPULATE_CADENCE_MS)
|
||||
.collect::<Vec<_>>(),
|
||||
primary_points,
|
||||
4,
|
||||
1,
|
||||
"ms_zero_offset",
|
||||
ms_zero_offset.clone(),
|
||||
ms_zero_offset,
|
||||
false,
|
||||
),
|
||||
(
|
||||
"one_field",
|
||||
(0..primary_points)
|
||||
.map(|point| point as i64 * RANGE_MANIPULATE_CADENCE_MS)
|
||||
.collect::<Vec<_>>(),
|
||||
primary_points,
|
||||
20,
|
||||
1,
|
||||
"ms_zero_offset_long_window",
|
||||
ms_zero_offset_long_window.clone(),
|
||||
ms_zero_offset_long_window,
|
||||
false,
|
||||
),
|
||||
(
|
||||
"one_field",
|
||||
(0..primary_points)
|
||||
.map(|point| point as i64 * RANGE_MANIPULATE_CADENCE_MS)
|
||||
.collect::<Vec<_>>(),
|
||||
primary_points,
|
||||
240,
|
||||
1,
|
||||
false,
|
||||
),
|
||||
(
|
||||
"regular_sparse",
|
||||
(0..sparse_evaluations)
|
||||
.step_by(2)
|
||||
.map(|point| point as i64 * RANGE_MANIPULATE_CADENCE_MS)
|
||||
.collect::<Vec<_>>(),
|
||||
sparse_evaluations,
|
||||
20,
|
||||
1,
|
||||
false,
|
||||
),
|
||||
(
|
||||
"all_empty_pathological",
|
||||
(0..primary_points)
|
||||
.map(|point| {
|
||||
(primary_points as i64 + 21 + point as i64) * RANGE_MANIPULATE_CADENCE_MS
|
||||
})
|
||||
.collect::<Vec<_>>(),
|
||||
primary_points,
|
||||
20,
|
||||
1,
|
||||
"ms_large_offset",
|
||||
ms_large_offset_narrow,
|
||||
ms_large_offset,
|
||||
true,
|
||||
),
|
||||
(
|
||||
"multi_field_control",
|
||||
(0..primary_points)
|
||||
.map(|point| point as i64 * RANGE_MANIPULATE_CADENCE_MS)
|
||||
.collect::<Vec<_>>(),
|
||||
primary_points,
|
||||
20,
|
||||
4,
|
||||
false,
|
||||
"ns_native_floor",
|
||||
nanosecond_floor_narrow,
|
||||
nanosecond_floor,
|
||||
true,
|
||||
),
|
||||
];
|
||||
|
||||
for (case_name, timestamps, evaluations, window_points, field_count, expect_empty) in
|
||||
cases.drain(..)
|
||||
{
|
||||
let expected_range_keys = (!expect_empty)
|
||||
.then(|| brute_force_range_keys(×tamps, evaluations, window_points));
|
||||
let input_batch = make_range_manipulate_batch(timestamps, field_count);
|
||||
let input_rows = input_batch.num_rows();
|
||||
let input_schema = input_batch.schema();
|
||||
let logical_input = LogicalPlan::EmptyRelation(EmptyRelation {
|
||||
produce_one_row: false,
|
||||
schema: input_schema.clone().to_dfschema_ref().unwrap(),
|
||||
});
|
||||
let field_columns = (0..field_count)
|
||||
.map(|field_index| format!("value_{field_index}"))
|
||||
.collect::<Vec<_>>();
|
||||
let logical_plan = RangeManipulate::new(
|
||||
0,
|
||||
(evaluations as i64 - 1) * RANGE_MANIPULATE_CADENCE_MS,
|
||||
RANGE_MANIPULATE_CADENCE_MS,
|
||||
0,
|
||||
window_points as i64 * RANGE_MANIPULATE_CADENCE_MS,
|
||||
"timestamp".to_string(),
|
||||
field_columns,
|
||||
logical_input,
|
||||
)
|
||||
.unwrap();
|
||||
let physical_input: Arc<dyn ExecutionPlan> = Arc::new(DataSourceExec::new(Arc::new(
|
||||
MemorySourceConfig::try_new(&[vec![input_batch]], input_schema, None).unwrap(),
|
||||
)));
|
||||
let execution_plan = logical_plan.to_execution_plan(physical_input);
|
||||
let runtime = tokio::runtime::Runtime::new().unwrap();
|
||||
let task_ctx = datafusion::prelude::SessionContext::new().task_ctx();
|
||||
let mut group = c.benchmark_group("cursor_scan_carriers");
|
||||
group.warm_up_time(std::time::Duration::from_millis(500));
|
||||
group.measurement_time(std::time::Duration::from_secs(2));
|
||||
group.sample_size(50);
|
||||
|
||||
// Validate the public operator output and RangeArray keys before timing.
|
||||
let output = runtime.block_on(execute_and_drain(execution_plan.clone(), task_ctx.clone()));
|
||||
validate_range_manipulate_output(&output, field_count, expected_range_keys.as_deref());
|
||||
for (case_name, narrow, wide, wide_falls_back) in cases {
|
||||
// Both arms must be timed on work the operator would really do: the narrow arm in
|
||||
// `i64`, and the wide arm in `i128` exactly when the narrow baselines overflow.
|
||||
let narrow_expected = narrow
|
||||
.bench_scan_narrow()
|
||||
.unwrap_or_else(|| panic!("{case_name}: the i64 arm must fit the narrow carrier"));
|
||||
assert_eq!(
|
||||
wide.bench_scan_narrow().is_none(),
|
||||
wide_falls_back,
|
||||
"{case_name}: the i128 arm's fallback trigger"
|
||||
);
|
||||
// Fairness check before timing: both carriers have to select the same ranges, which
|
||||
// makes the sample and window comparisons of the two arms identical.
|
||||
assert_eq!(
|
||||
wide.bench_scan_wide(),
|
||||
narrow_expected,
|
||||
"{case_name}: the two carriers must select the same ranges"
|
||||
);
|
||||
assert_eq!(
|
||||
narrow_expected.len(),
|
||||
narrow.evaluations(),
|
||||
"{case_name}: one range per aligned evaluation instant"
|
||||
);
|
||||
|
||||
// This measures public RangeManipulate execution wall time, including stream draining.
|
||||
group.throughput(criterion::Throughput::Elements(
|
||||
narrow.timestamps.len() as u64
|
||||
));
|
||||
group.bench_with_input(
|
||||
BenchmarkId::new(
|
||||
case_name,
|
||||
format!("N{input_rows}_eval{evaluations}_window{window_points}x15s"),
|
||||
),
|
||||
&(),
|
||||
|b, _| {
|
||||
b.iter(|| {
|
||||
let output = runtime
|
||||
.block_on(execute_and_drain(execution_plan.clone(), task_ctx.clone()));
|
||||
std::hint::black_box(output);
|
||||
})
|
||||
BenchmarkId::new(format!("{case_name}/i64"), narrow.label()),
|
||||
&narrow,
|
||||
|b, workload| {
|
||||
b.iter(|| std::hint::black_box(workload.bench_scan_narrow()));
|
||||
},
|
||||
);
|
||||
group.bench_with_input(
|
||||
BenchmarkId::new(format!("{case_name}/i128"), wide.label()),
|
||||
&wide,
|
||||
|b, workload| {
|
||||
b.iter(|| std::hint::black_box(workload.bench_scan_wide()));
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -1196,5 +1256,5 @@ criterion_group!(
|
||||
bench_rate_window_steps,
|
||||
bench_edge_count_functions,
|
||||
bench_extrema_functions,
|
||||
bench_range_manipulate_wall_time
|
||||
bench_cursor_scan_carriers
|
||||
);
|
||||
|
||||
@@ -46,6 +46,7 @@ use prost::Message;
|
||||
use snafu::ResultExt;
|
||||
|
||||
use crate::error::{DeserializeSnafu, Result};
|
||||
use crate::extension_plan::range_manipulate::{Nanoseconds, ScanNanoseconds, narrow_scan};
|
||||
use crate::extension_plan::series_divide::SeriesDivide;
|
||||
use crate::extension_plan::{
|
||||
METRIC_NUM_SERIES, Millisecond, is_prometheus_stale_sample, local_offset,
|
||||
@@ -691,6 +692,7 @@ impl InstantManipulateStream {
|
||||
let (timestamps, _) = timestamp_array_to_primitive(ts_column).ok_or_else(|| {
|
||||
DataFusionError::Execution("Time index column is not a timestamp".into())
|
||||
})?;
|
||||
let timestamp_null_count = timestamps.null_count();
|
||||
let timestamps = timestamps.values();
|
||||
let len = timestamps.len();
|
||||
// Shift the native-tick timeline in i128 before comparing samples. Doing
|
||||
@@ -735,28 +737,77 @@ impl InstantManipulateStream {
|
||||
let estimated_points = estimated_points as usize;
|
||||
let aligned_start = aligned_start as i64;
|
||||
let aligned_end = aligned_end as i64;
|
||||
let mut take_indices = Vec::with_capacity(estimated_points);
|
||||
let mut aligned_ts = Vec::with_capacity(estimated_points);
|
||||
let mut cursor = 0;
|
||||
for expected_ms in (aligned_start..=aligned_end).step_by(self.interval as usize) {
|
||||
let expected = (expected_ms as i128) * 1_000_000;
|
||||
// Narrow `i64` nanoseconds whenever the batch fits them, exact `i128` otherwise.
|
||||
let narrow = (timestamp_null_count == 0)
|
||||
.then(|| {
|
||||
narrow_scan(
|
||||
self.time_unit,
|
||||
self.offset,
|
||||
self.lookback_delta,
|
||||
timestamps,
|
||||
aligned_start,
|
||||
aligned_end,
|
||||
)
|
||||
})
|
||||
.flatten();
|
||||
if let Some(scan) = &narrow {
|
||||
debug_assert_eq!(i64::try_from(first_ns), Ok(scan.shift(timestamps[0])));
|
||||
debug_assert_eq!(i64::try_from(last_ns), Ok(scan.shift(timestamps[len - 1])));
|
||||
}
|
||||
let (take_indices, aligned_ts) = match narrow {
|
||||
Some(scan) => self.scan_instants::<i64>(
|
||||
timestamps,
|
||||
aligned_start,
|
||||
aligned_end,
|
||||
&scan,
|
||||
estimated_points,
|
||||
&is_stale,
|
||||
),
|
||||
None => self.scan_instants::<i128>(
|
||||
timestamps,
|
||||
aligned_start,
|
||||
aligned_end,
|
||||
&ScanNanoseconds::<i128>::wide(self.time_unit, self.offset, self.lookback_delta),
|
||||
estimated_points,
|
||||
&is_stale,
|
||||
),
|
||||
};
|
||||
self.take_record_batch_optional(input, take_indices, aligned_ts)
|
||||
}
|
||||
|
||||
/// Chooses the sample of every aligned evaluation instant, in `Ns` nanoseconds.
|
||||
///
|
||||
/// Keeps the first row among exact timestamp ties and the latest preceding row
|
||||
/// otherwise; zero lookback admits only exact samples. Staleness is tested after
|
||||
/// choosing: a selected stale marker suppresses this evaluation rather than
|
||||
/// falling back to an older finite value.
|
||||
fn scan_instants<Ns: Nanoseconds>(
|
||||
&self,
|
||||
timestamps: &[i64],
|
||||
start: Millisecond,
|
||||
end: Millisecond,
|
||||
scan: &ScanNanoseconds<Ns>,
|
||||
capacity: usize,
|
||||
is_stale: impl Fn(usize) -> bool,
|
||||
) -> (Vec<u64>, Vec<Millisecond>) {
|
||||
let len = timestamps.len();
|
||||
let mut cursor = 0usize;
|
||||
let mut take_indices = Vec::with_capacity(capacity);
|
||||
let mut aligned_ts = Vec::with_capacity(capacity);
|
||||
for expected_ms in (start..=end).step_by(self.interval as usize) {
|
||||
let expected = scan.instant(expected_ms);
|
||||
let mut exact_candidate = None;
|
||||
while cursor < len && to_nanoseconds(timestamps[cursor]) <= expected {
|
||||
if to_nanoseconds(timestamps[cursor]) == expected && exact_candidate.is_none() {
|
||||
while cursor < len && scan.shift(timestamps[cursor]) <= expected {
|
||||
if scan.shift(timestamps[cursor]) == expected && exact_candidate.is_none() {
|
||||
exact_candidate = Some(cursor);
|
||||
}
|
||||
cursor += 1;
|
||||
}
|
||||
// Keep the first row among exact timestamp ties; otherwise use the
|
||||
// latest preceding row. Zero lookback admits only exact samples.
|
||||
// Test staleness after choosing: a selected stale marker suppresses
|
||||
// this evaluation rather than falling back to an older finite value.
|
||||
let Some(candidate) = exact_candidate.or_else(|| cursor.checked_sub(1)) else {
|
||||
continue;
|
||||
};
|
||||
let candidate_ts = to_nanoseconds(timestamps[candidate]);
|
||||
let lower = expected - (self.lookback_delta as i128) * 1_000_000;
|
||||
if (candidate_ts == expected || candidate_ts > lower)
|
||||
let candidate_ts = scan.shift(timestamps[candidate]);
|
||||
if (candidate_ts == expected || candidate_ts > scan.window_start(expected))
|
||||
&& candidate_ts <= expected
|
||||
&& !is_stale(candidate)
|
||||
{
|
||||
@@ -764,7 +815,7 @@ impl InstantManipulateStream {
|
||||
aligned_ts.push(expected_ms);
|
||||
}
|
||||
}
|
||||
self.take_record_batch_optional(input, take_indices, aligned_ts)
|
||||
(take_indices, aligned_ts)
|
||||
}
|
||||
|
||||
/// Helper function to apply "take" on record batch.
|
||||
@@ -1775,6 +1826,81 @@ mod test {
|
||||
.await;
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nullable_interior_timestamp_matches_wide_scan() {
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
Field::new(
|
||||
TIME_INDEX_COLUMN,
|
||||
DataType::Timestamp(TimeUnit::Second, None),
|
||||
true,
|
||||
),
|
||||
Field::new("value", DataType::Float64, false),
|
||||
]));
|
||||
let timestamps = [0, i64::MAX, 1];
|
||||
let array = TimestampSecondArray::new(
|
||||
timestamps.to_vec().into(),
|
||||
Some(NullBuffer::from(vec![false, false, true])),
|
||||
);
|
||||
let input = RecordBatch::try_new(
|
||||
schema.clone(),
|
||||
vec![
|
||||
Arc::new(array),
|
||||
Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0])),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
let stream = InstantManipulateStream {
|
||||
offset: 0,
|
||||
start: 0,
|
||||
end: 1_000,
|
||||
lookback_delta: 1_000,
|
||||
interval: 1_000,
|
||||
time_index: 0,
|
||||
time_unit: TimeUnit::Second,
|
||||
field_indices: [Some(1), None],
|
||||
tsid_index: None,
|
||||
reuse_tsid_column: false,
|
||||
schema: Arc::new(Schema::new(vec![
|
||||
Field::new(
|
||||
TIME_INDEX_COLUMN,
|
||||
DataType::Timestamp(TimeUnit::Millisecond, None),
|
||||
false,
|
||||
),
|
||||
Field::new("value", DataType::Float64, false),
|
||||
])),
|
||||
input: Box::pin(
|
||||
datafusion::physical_plan::memory::MemoryStream::try_new(vec![], schema, None)
|
||||
.unwrap(),
|
||||
),
|
||||
metric: BaselineMetrics::new(&ExecutionPlanMetricsSet::new(), 0),
|
||||
num_series: Count::new(),
|
||||
};
|
||||
|
||||
let output = stream.manipulate(input).unwrap();
|
||||
let output_timestamps = output
|
||||
.column(0)
|
||||
.as_any()
|
||||
.downcast_ref::<TimestampMillisecondArray>()
|
||||
.unwrap();
|
||||
let output_values = output
|
||||
.column(1)
|
||||
.as_any()
|
||||
.downcast_ref::<Float64Array>()
|
||||
.unwrap();
|
||||
let wide = stream.scan_instants::<i128>(
|
||||
×tamps,
|
||||
0,
|
||||
1_000,
|
||||
&ScanNanoseconds::<i128>::wide(TimeUnit::Second, 0, 1_000),
|
||||
2,
|
||||
|_| false,
|
||||
);
|
||||
assert_eq!(output_timestamps.values(), &[0]);
|
||||
assert_eq!(output_values.values(), &[1.0]);
|
||||
assert_eq!(wide, (vec![0], vec![0]));
|
||||
assert_eq!(wide.1, output_timestamps.values().to_vec());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn exact_ties_select_first_and_lookback_uses_latest() {
|
||||
for (values, expected_timestamp) in [
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,119 @@
|
||||
# PromQL long-window rate benchmark for issue #9070.
|
||||
#
|
||||
# 2048 series × 1440 samples at 15-second cadence = 2,949,120 rows spanning the
|
||||
# six hours from 2024-01-01T00:00:00Z to 2024-01-01T06:00:00Z. Evaluation covers
|
||||
# the last two hours of that data (04:00:00Z–06:00:00Z at a 15-second step, 481
|
||||
# evaluations), so every evaluation holds a full 240-sample one-hour window.
|
||||
# `rate_1h` is the primary regressed shape: it drives the RangeManipulate
|
||||
# timestamp payload and the cursor scan across a long, heavily overlapping
|
||||
# window set. `rate_5m` is the small-window control, `offset 1h` exercises the
|
||||
# shifted payload path, and the bare selector controls unrelated-path variance.
|
||||
#
|
||||
# The generator derives each tag value as `series % cardinality`, so the number
|
||||
# of distinct (host, instance) series is the least common multiple of the two
|
||||
# cardinalities. `instance` declares 2048 values so the host(16) × instance tag
|
||||
# shape still yields 2048 distinct series; a 128-value instance tag would cap
|
||||
# this fixture at 128 distinct series.
|
||||
#
|
||||
# The deterministic wave is not a true monotonic counter, so the rate queries
|
||||
# measure latency on the regular rate path rather than a numeric counter-rate
|
||||
# oracle.
|
||||
|
||||
[case]
|
||||
name = "promql_long_range_rate"
|
||||
description = "PromQL long-window rate latency regression for #9070 over six hours of 15-second samples"
|
||||
issue = "https://github.com/GreptimeTeam/greptimedb/issues/9070"
|
||||
|
||||
[scenario]
|
||||
kind = "direct_readable_sst"
|
||||
seed = 9070
|
||||
|
||||
[[scenario.tables]]
|
||||
database = "public"
|
||||
name = "promql_long_range_rate"
|
||||
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 = 2048, prefix = "instance" }
|
||||
|
||||
[[scenario.tables.columns]]
|
||||
name = "value"
|
||||
type = "DOUBLE"
|
||||
semantic = "field"
|
||||
distribution = { kind = "deterministic_wave", min = 0.0, max = 1000.0 }
|
||||
|
||||
[[scenario.tables.columns]]
|
||||
name = "ts"
|
||||
type = "TIMESTAMP(9)"
|
||||
semantic = "timestamp"
|
||||
|
||||
# Eight 45-minute SSTs (180 timestamps each) tile the six-hour range.
|
||||
# A (03:00:00Z, 06:00:00Z] scan prunes to four of them; the offset query scans
|
||||
# the five SSTs covering [02:00:00Z, 05:00:00Z).
|
||||
[scenario.layout]
|
||||
regions = 1
|
||||
sst_count = 8
|
||||
rows_per_sst = 368640
|
||||
row_group_size = 36864
|
||||
series_count = 2048
|
||||
start_unix_nanos = 1704067200000000000
|
||||
step_nanos = 15000000000
|
||||
time_range_layout = "non_overlapping_per_sst"
|
||||
series_layout = "timestamp_major"
|
||||
|
||||
# 240 samples per full window: the primary regressed shape.
|
||||
[[scenario.queries]]
|
||||
name = "rate_1h"
|
||||
kind = "tql"
|
||||
query = "TQL ANALYZE VERBOSE (1704081600, 1704088800, '15s') rate(promql_long_range_rate{host=~'host.*'}[1h])"
|
||||
warmup = 3
|
||||
iterations = 9
|
||||
|
||||
[scenario.queries.thresholds]
|
||||
max_candidate_latency_regression_pct = 20
|
||||
|
||||
# Twenty samples per full window: small-window control.
|
||||
[[scenario.queries]]
|
||||
name = "rate_5m"
|
||||
kind = "tql"
|
||||
query = "TQL ANALYZE VERBOSE (1704081600, 1704088800, '15s') rate(promql_long_range_rate{host=~'host.*'}[5m])"
|
||||
warmup = 3
|
||||
iterations = 9
|
||||
|
||||
[scenario.queries.thresholds]
|
||||
max_candidate_latency_regression_pct = 20
|
||||
|
||||
# The same long window shifted back an hour: non-zero-offset payload path.
|
||||
[[scenario.queries]]
|
||||
name = "rate_1h_offset_1h"
|
||||
kind = "tql"
|
||||
query = "TQL ANALYZE VERBOSE (1704081600, 1704088800, '15s') rate(promql_long_range_rate{host=~'host.*'}[1h] offset 1h)"
|
||||
warmup = 3
|
||||
iterations = 9
|
||||
|
||||
[scenario.queries.thresholds]
|
||||
max_candidate_latency_regression_pct = 20
|
||||
|
||||
# Bare selector over the same evaluation range: unrelated-path variance control.
|
||||
[[scenario.queries]]
|
||||
name = "plain_selector_control"
|
||||
kind = "tql"
|
||||
query = "TQL ANALYZE VERBOSE (1704081600, 1704088800, '15s') promql_long_range_rate{host=~'host.*'}"
|
||||
warmup = 3
|
||||
iterations = 9
|
||||
|
||||
[scenario.queries.thresholds]
|
||||
max_candidate_latency_regression_pct = 20
|
||||
Reference in New Issue
Block a user