diff --git a/asap-query-engine/Cargo.toml b/asap-query-engine/Cargo.toml index bb7a8d4..df58fcc 100644 --- a/asap-query-engine/Cargo.toml +++ b/asap-query-engine/Cargo.toml @@ -84,11 +84,6 @@ path = "src/bin/e2e_quickstart_resource_test.rs" tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } -[[bench]] -name = "simple_store_bench" -harness = false -required-features = ["legacy_stores"] - [features] #default = ["lock_profiling", "extra_debugging"] default = ["jemalloc"] @@ -98,8 +93,3 @@ jemalloc = ["dep:tikv-jemallocator"] lock_profiling = [] # Enable extra debugging output extra_debugging = [] -# Legacy SimpleMapStore implementations, kept only for the simple_store_bench -# comparison benchmark. Off by default so the Store trait's real implementors -# are just SimpleMapStoreGlobal/PerKey — adding a Store trait method doesn't -# require updating the legacy variants. -legacy_stores = [] diff --git a/asap-query-engine/benches/simple_store_bench.rs b/asap-query-engine/benches/simple_store_bench.rs deleted file mode 100644 index 831f019..0000000 --- a/asap-query-engine/benches/simple_store_bench.rs +++ /dev/null @@ -1,858 +0,0 @@ -//! Benchmarks for `LegacySimpleMapStore` — insert, range query, exact query, -//! store-analyze, and concurrent reads. -//! -//! These benchmarks profile the legacy store implementation -//! (`LegacySimpleMapStoreGlobal` / `LegacySimpleMapStorePerKey`) and provide -//! concrete measurements of algorithm complexity for: -//! -//! | Operation | Expected complexity | -//! |------------------------------------|--------------------------| -//! | `insert_precomputed_output_batch` | O(B) | -//! | `query_precomputed_output` (range) | O(W·log W + k) | -//! | `query_precomputed_output_exact` | O(1) HashMap lookup | -//! | `get_earliest_timestamp` (analyze) | O(A) — scan agg-id map | -//! | concurrent reads (n threads) | serialised by write lock | -//! -//! where B = batch size, W = stored windows, k = result entries, A = agg IDs. -//! -//! Two accumulator types are benchmarked: -//! - `sum` — `SumAccumulator` (trivial f64, ~0 clone cost, baseline) -//! - `kll` — `DatasketchesKLLAccumulator` k=200 (~1 KB sketch, realistic clone cost) -//! -//! Run with: -//! cargo bench -p query_engine_rust --bench simple_store_bench -//! -//! Results land in `target/criterion/`. - -use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion, Throughput}; -use promql_utilities::data_model::KeyByLabelNames; -use query_engine_rust::data_model::{ - AggregateCore, AggregationType, CleanupPolicy, KeyByLabelValues, LockStrategy, StreamingConfig, - WindowType, -}; -use query_engine_rust::precompute_operators::{DatasketchesKLLAccumulator, SumAccumulator}; -use query_engine_rust::stores::simple_map_store::legacy::{ - LegacySimpleMapStoreGlobal, LegacySimpleMapStorePerKey, -}; -use query_engine_rust::stores::Store; -use query_engine_rust::{AggregationConfig, PrecomputedOutput, SimpleMapStore}; -use std::collections::HashMap; -use std::sync::{Arc, Barrier}; - -#[derive(Clone, Copy)] -enum StoreKind { - LegacyPerKey, - LegacyGlobal, - CurrentPerKey, - CurrentGlobal, -} - -impl StoreKind { - const ALL: [Self; 4] = [ - Self::LegacyPerKey, - Self::LegacyGlobal, - Self::CurrentPerKey, - Self::CurrentGlobal, - ]; - - fn slug(self) -> &'static str { - match self { - Self::LegacyPerKey => "legacy/per_key", - Self::LegacyGlobal => "legacy/global", - Self::CurrentPerKey => "current/per_key", - Self::CurrentGlobal => "current/global", - } - } - - fn build(self, config: Arc, cleanup_policy: CleanupPolicy) -> Arc { - match self { - Self::LegacyPerKey => Arc::new(LegacySimpleMapStorePerKey::new(config, cleanup_policy)), - Self::LegacyGlobal => Arc::new(LegacySimpleMapStoreGlobal::new(config, cleanup_policy)), - Self::CurrentPerKey => Arc::new(SimpleMapStore::new_with_strategy( - config, - cleanup_policy, - LockStrategy::PerKey, - )), - Self::CurrentGlobal => Arc::new(SimpleMapStore::new_with_strategy( - config, - cleanup_policy, - LockStrategy::Global, - )), - } - } -} - -#[derive(Clone, Copy)] -enum AccumulatorKind { - Sum, - Kll, -} - -impl AccumulatorKind { - const ALL: [Self; 2] = [Self::Sum, Self::Kll]; - - fn slug(self) -> &'static str { - match self { - Self::Sum => "sum", - Self::Kll => "kll", - } - } - - fn aggregation_type(self) -> AggregationType { - match self { - Self::Sum => AggregationType::Sum, - Self::Kll => AggregationType::DatasketchesKLL, - } - } - - fn build(self, value: f64) -> Box { - match self { - Self::Sum => Box::new(SumAccumulator::with_sum(value)), - Self::Kll => { - let mut acc = DatasketchesKLLAccumulator::new(200); - for v in 0..20 { - acc.update(v as f64 * (value + 1.0)); - } - Box::new(acc) - } - } - } -} - -fn make_agg_config( - agg_id: u64, - aggregation_type: AggregationType, - metric: &str, - num_aggregates_to_retain: Option, - read_count_threshold: Option, -) -> AggregationConfig { - AggregationConfig::new( - agg_id, - aggregation_type, - "".to_string(), - HashMap::new(), - KeyByLabelNames::empty(), - KeyByLabelNames::empty(), - KeyByLabelNames::empty(), - "".to_string(), - 60_000, // window_size_ms - 60_000, // slide_interval_ms - WindowType::Tumbling, // window_type - "".to_string(), // spatial_filter - metric.to_string(), - num_aggregates_to_retain, - read_count_threshold, - None, // table_name - None, // value_column - ) -} - -fn make_streaming_config( - agg_ids: &[u64], - accumulator_kind: AccumulatorKind, - metric: &str, - num_aggregates_to_retain: Option, - read_count_threshold: Option, -) -> Arc { - let configs = agg_ids - .iter() - .copied() - .map(|agg_id| { - ( - agg_id, - make_agg_config( - agg_id, - accumulator_kind.aggregation_type(), - metric, - num_aggregates_to_retain, - read_count_threshold, - ), - ) - }) - .collect(); - Arc::new(StreamingConfig::new(configs)) -} - -fn make_batch( - count: usize, - agg_id: u64, - base_ts: u64, - window_ms: u64, - accumulator_kind: AccumulatorKind, -) -> Vec<(PrecomputedOutput, Box)> { - (0..count as u64) - .map(|i| { - let start = base_ts + i * window_ms; - let end = start + window_ms; - ( - PrecomputedOutput::new(start, end, None, agg_id), - accumulator_kind.build(i as f64), - ) - }) - .collect() -} - -fn insert_labelled_entry( - store: &dyn Store, - start: u64, - end: u64, - label: &str, - agg_id: u64, - accumulator_kind: AccumulatorKind, - value: f64, -) { - let output = PrecomputedOutput::new( - start, - end, - Some(KeyByLabelValues::new_with_labels(vec![label.to_string()])), - agg_id, - ); - store - .insert_precomputed_output(output, accumulator_kind.build(value)) - .unwrap(); -} - -fn populate_store_labelled( - store: &dyn Store, - time_ranges: usize, - labels: usize, - agg_id: u64, - accumulator_kind: AccumulatorKind, -) { - for i in 0..time_ranges { - let start = i as u64 * 1_000; - let end = start + 1_000; - for j in 0..labels { - insert_labelled_entry( - store, - start, - end, - &format!("host-{j}"), - agg_id, - accumulator_kind, - 1.0, - ); - } - } -} - -fn populate_store_batch( - store: &dyn Store, - num_windows: usize, - agg_id: u64, - accumulator_kind: AccumulatorKind, -) { - let batch = make_batch(num_windows, agg_id, 1_000_000, 60_000, accumulator_kind); - store.insert_precomputed_output_batch(batch).unwrap(); -} - -fn build_populated_store( - kind: StoreKind, - accumulator_kind: AccumulatorKind, - time_ranges: usize, - labels: usize, -) -> Arc { - let config = make_streaming_config(&[1], accumulator_kind, "test_metric", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - populate_store_labelled(store.as_ref(), time_ranges, labels, 1, accumulator_kind); - store -} - -fn bench_insert_batch_size(c: &mut Criterion) { - let mut group = c.benchmark_group("insert/batch_size"); - - for &batch_size in &[100usize, 1_000, 5_000, 10_000] { - group.throughput(Throughput::Elements(batch_size as u64)); - - for kind in StoreKind::ALL { - for accumulator_kind in AccumulatorKind::ALL { - let id = format!("{}/{}", kind.slug(), accumulator_kind.slug()); - group.bench_with_input(BenchmarkId::new(id, batch_size), &batch_size, |b, &n| { - b.iter_batched( - || { - let config = make_streaming_config( - &[1], - accumulator_kind, - "cpu_usage", - None, - None, - ); - ( - kind.build(config, CleanupPolicy::NoCleanup), - make_batch(n, 1, 1_000_000, 60_000, accumulator_kind), - ) - }, - |(store, batch)| { - store.insert_precomputed_output_batch(batch).unwrap(); - }, - criterion::BatchSize::SmallInput, - ); - }); - } - } - } - - group.finish(); -} - -fn bench_insert_num_agg_ids(c: &mut Criterion) { - let mut group = c.benchmark_group("insert/num_agg_ids"); - const TOTAL_ITEMS: usize = 1_000; - - for &num_ids in &[1usize, 10, 50, 200] { - group.throughput(Throughput::Elements(TOTAL_ITEMS as u64)); - - for kind in StoreKind::ALL { - group.bench_with_input(BenchmarkId::new(kind.slug(), num_ids), &num_ids, |b, &n| { - b.iter_batched( - || { - let agg_ids: Vec = (1..=n as u64).collect(); - let config = make_streaming_config( - &agg_ids, - AccumulatorKind::Sum, - "cpu_usage", - None, - None, - ); - let store = kind.build(config, CleanupPolicy::NoCleanup); - let per_id = TOTAL_ITEMS / n; - let mut batch = Vec::with_capacity(per_id * n); - for agg_id in agg_ids { - batch.extend(make_batch( - per_id, - agg_id, - 1_000_000, - 60_000, - AccumulatorKind::Sum, - )); - } - (store, batch) - }, - |(store, batch)| { - store.insert_precomputed_output_batch(batch).unwrap(); - }, - criterion::BatchSize::SmallInput, - ); - }); - } - } - - group.finish(); -} - -fn bench_query_range_store_size(c: &mut Criterion) { - let mut group = c.benchmark_group("query/range_store_size"); - - for &num_windows in &[500usize, 1_000, 5_000, 10_000] { - for kind in StoreKind::ALL { - for accumulator_kind in AccumulatorKind::ALL { - let store = { - let config = - make_streaming_config(&[1], accumulator_kind, "cpu_usage", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - populate_store_batch(store.as_ref(), num_windows, 1, accumulator_kind); - store - }; - let id = format!("{}/{}", kind.slug(), accumulator_kind.slug()); - let query_start = 1_000_000u64; - let query_end = query_start + num_windows as u64 * 60_000; - - group.bench_with_input(BenchmarkId::new(id, num_windows), &num_windows, |b, _| { - b.iter(|| { - black_box( - store - .query_precomputed_output("cpu_usage", 1, query_start, query_end) - .unwrap(), - ) - }); - }); - } - } - } - - group.finish(); -} - -fn bench_query_exact_store_size(c: &mut Criterion) { - let mut group = c.benchmark_group("query/exact_store_size"); - - for &num_windows in &[500usize, 1_000, 5_000, 10_000] { - for kind in StoreKind::ALL { - let store = { - let config = - make_streaming_config(&[1], AccumulatorKind::Sum, "cpu_usage", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - populate_store_batch(store.as_ref(), num_windows, 1, AccumulatorKind::Sum); - store - }; - let exact_start = 1_000_000u64 + (num_windows as u64 - 1) * 60_000; - let exact_end = exact_start + 60_000; - - group.bench_with_input( - BenchmarkId::new(kind.slug(), num_windows), - &num_windows, - |b, _| { - b.iter(|| { - black_box( - store - .query_precomputed_output_exact( - "cpu_usage", - 1, - exact_start, - exact_end, - ) - .unwrap(), - ) - }); - }, - ); - } - } - - group.finish(); -} - -fn bench_store_analyze(c: &mut Criterion) { - let mut group = c.benchmark_group("store_analyze/num_agg_ids"); - - for &num_ids in &[10usize, 100, 500, 1_000] { - let agg_ids: Vec = (1..=num_ids as u64).collect(); - for kind in StoreKind::ALL { - let config = - make_streaming_config(&agg_ids, AccumulatorKind::Sum, "cpu_usage", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - for agg_id in 1..=num_ids as u64 { - let output = PrecomputedOutput::new(1_000_000, 1_060_000, None, agg_id); - store - .insert_precomputed_output(output, AccumulatorKind::Sum.build(1.0)) - .unwrap(); - } - - group.bench_with_input(BenchmarkId::new(kind.slug(), num_ids), &num_ids, |b, _| { - b.iter(|| black_box(store.get_earliest_timestamp_per_aggregation_id().unwrap())); - }); - } - } - - group.finish(); -} - -fn bench_concurrent_reads(c: &mut Criterion) { - let mut group = c.benchmark_group("concurrent_reads/thread_count"); - let num_windows = 5_000usize; - let query_start = 1_000_000u64; - let query_end = query_start + num_windows as u64 * 60_000; - - for kind in StoreKind::ALL { - let config = make_streaming_config(&[1], AccumulatorKind::Sum, "cpu_usage", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - populate_store_batch(store.as_ref(), num_windows, 1, AccumulatorKind::Sum); - - for &num_threads in &[1usize, 2, 4, 8] { - group.bench_with_input( - BenchmarkId::new(kind.slug(), num_threads), - &num_threads, - |b, &n| { - let store = store.clone(); - b.iter(|| { - let handles: Vec<_> = (0..n) - .map(|_| { - let store = store.clone(); - std::thread::spawn(move || { - store - .query_precomputed_output( - "cpu_usage", - 1, - query_start, - query_end, - ) - .unwrap() - }) - }) - .collect(); - for handle in handles { - black_box(handle.join().unwrap()); - } - }); - }, - ); - } - } - - group.finish(); -} - -fn bench_concurrent_writes(c: &mut Criterion) { - let mut group = c.benchmark_group("concurrent_writes/thread_count"); - let labels = 10usize; - let entries_per_thread = 500usize; - let time_ranges_per_thread = entries_per_thread / labels; - - for kind in StoreKind::ALL { - for &num_threads in &[1usize, 2, 4, 8] { - group.bench_with_input( - BenchmarkId::new(kind.slug(), num_threads), - &num_threads, - |b, &n| { - b.iter(|| { - let config = make_streaming_config( - &[1], - AccumulatorKind::Sum, - "test_metric", - None, - None, - ); - let store = kind.build(config, CleanupPolicy::NoCleanup); - let barrier = Arc::new(Barrier::new(n)); - std::thread::scope(|scope| { - for t in 0..n { - let store = store.clone(); - let barrier = barrier.clone(); - scope.spawn(move || { - barrier.wait(); - for i in 0..time_ranges_per_thread { - let start = i as u64 * 1_000; - let end = start + 1_000; - for j in 0..labels { - insert_labelled_entry( - store.as_ref(), - start, - end, - &format!("thread-{t}-host-{j}"), - 1, - AccumulatorKind::Sum, - 1.0, - ); - } - } - }); - } - }); - black_box(store); - }); - }, - ); - } - } - - group.finish(); -} - -fn bench_concurrent_mixed_read_write(c: &mut Criterion) { - let mut group = c.benchmark_group("concurrent_mixed_rw/config"); - let writers = 2usize; - let readers = 2usize; - let labels = 10usize; - let time_ranges = 1_000usize; - let total_threads = writers + readers; - - for kind in StoreKind::ALL { - let store = build_populated_store(kind, AccumulatorKind::Sum, time_ranges, labels); - let query_end = time_ranges as u64 * 1_000 / 10; - - group.bench_function(kind.slug(), |b| { - let store = store.clone(); - b.iter(|| { - let barrier = Arc::new(Barrier::new(total_threads)); - std::thread::scope(|scope| { - for writer_id in 0..writers { - let store = store.clone(); - let barrier = barrier.clone(); - scope.spawn(move || { - barrier.wait(); - for offset in 0..50usize { - let start = (time_ranges + writer_id * 50 + offset) as u64 * 1_000; - let end = start + 1_000; - for label_id in 0..labels { - insert_labelled_entry( - store.as_ref(), - start, - end, - &format!("mixed-{writer_id}-host-{label_id}"), - 1, - AccumulatorKind::Sum, - 1.0, - ); - } - } - }); - } - - for _ in 0..readers { - let store = store.clone(); - let barrier = barrier.clone(); - scope.spawn(move || { - barrier.wait(); - for _ in 0..20 { - black_box( - store - .query_precomputed_output("test_metric", 1, 0, query_end) - .unwrap(), - ); - } - }); - } - }); - }); - }); - } - - group.finish(); -} - -fn bench_cleanup_overhead(c: &mut Criterion) { - let mut group = c.benchmark_group("cleanup_overhead"); - let labels = 5usize; - - for kind in StoreKind::ALL { - group.bench_function(format!("{}/no_cleanup", kind.slug()), |b| { - b.iter(|| { - let config = - make_streaming_config(&[1], AccumulatorKind::Sum, "test_metric", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - populate_store_labelled(store.as_ref(), 200, labels, 1, AccumulatorKind::Sum); - black_box(store); - }); - }); - - group.bench_function(format!("{}/circular_buffer", kind.slug()), |b| { - b.iter(|| { - let config = make_streaming_config( - &[1], - AccumulatorKind::Sum, - "test_metric", - Some(50), - None, - ); - let store = kind.build(config, CleanupPolicy::CircularBuffer); - populate_store_labelled(store.as_ref(), 200, labels, 1, AccumulatorKind::Sum); - black_box(store); - }); - }); - - group.bench_function(format!("{}/read_based", kind.slug()), |b| { - b.iter(|| { - let config = - make_streaming_config(&[1], AccumulatorKind::Sum, "test_metric", None, Some(2)); - let store = kind.build(config, CleanupPolicy::ReadBased); - populate_store_labelled(store.as_ref(), 100, labels, 1, AccumulatorKind::Sum); - - for _ in 0..2 { - black_box( - store - .query_precomputed_output("test_metric", 1, 0, 100_000) - .unwrap(), - ); - } - - for i in 100..200usize { - let start = i as u64 * 1_000; - let end = start + 1_000; - for j in 0..labels { - insert_labelled_entry( - store.as_ref(), - start, - end, - &format!("host-{j}"), - 1, - AccumulatorKind::Sum, - 1.0, - ); - } - } - - black_box(store); - }); - }); - } - - group.finish(); -} - -fn bench_query_patterns(c: &mut Criterion) { - let mut group = c.benchmark_group("query_patterns"); - let time_ranges = 1_000usize; - let labels = 10usize; - let total_time = time_ranges as u64 * 1_000; - - for kind in StoreKind::ALL { - let store = build_populated_store(kind, AccumulatorKind::Sum, time_ranges, labels); - - for (name, start, end) in [ - ("full_scan", 0, total_time), - ("wide_50pct", 0, total_time / 2), - ("narrow_1pct", 0, total_time / 100), - ("miss", total_time + 1_000_000, total_time + 1_001_000), - ] { - group.bench_function(format!("{}/{}", kind.slug(), name), |b| { - let store = store.clone(); - b.iter(|| { - black_box( - store - .query_precomputed_output("test_metric", 1, start, end) - .unwrap(), - ); - }); - }); - } - } - - group.finish(); -} - -fn bench_high_label_cardinality(c: &mut Criterion) { - let mut group = c.benchmark_group("high_label_cardinality"); - let time_ranges = 20usize; - - for &label_count in &[10usize, 100, 500, 1_000] { - for kind in StoreKind::ALL { - group.bench_with_input( - BenchmarkId::new(format!("{}/insert", kind.slug()), label_count), - &label_count, - |b, &lc| { - b.iter(|| { - let store = - build_populated_store(kind, AccumulatorKind::Sum, time_ranges, lc); - black_box(store); - }); - }, - ); - - let store = build_populated_store(kind, AccumulatorKind::Sum, time_ranges, label_count); - let query_end = time_ranges as u64 * 1_000; - group.bench_with_input( - BenchmarkId::new(format!("{}/query", kind.slug()), label_count), - &label_count, - |b, _| { - let store = store.clone(); - b.iter(|| { - black_box( - store - .query_precomputed_output("test_metric", 1, 0, query_end) - .unwrap(), - ); - }); - }, - ); - } - } - - group.finish(); -} - -fn bench_multi_agg_id(c: &mut Criterion) { - let mut group = c.benchmark_group("multi_agg_id"); - let agg_ids: Vec = (1..=10).collect(); - let time_ranges = 100usize; - let labels = 5usize; - - for kind in StoreKind::ALL { - group.bench_function(format!("{}/insert_10_agg_ids", kind.slug()), |b| { - b.iter(|| { - let config = make_streaming_config( - &agg_ids, - AccumulatorKind::Sum, - "test_metric", - None, - None, - ); - let store = kind.build(config, CleanupPolicy::NoCleanup); - for &agg_id in &agg_ids { - populate_store_labelled( - store.as_ref(), - time_ranges, - labels, - agg_id, - AccumulatorKind::Sum, - ); - } - black_box(store); - }); - }); - - let config = - make_streaming_config(&agg_ids, AccumulatorKind::Sum, "test_metric", None, None); - let store = kind.build(config, CleanupPolicy::NoCleanup); - for &agg_id in &agg_ids { - populate_store_labelled( - store.as_ref(), - time_ranges, - labels, - agg_id, - AccumulatorKind::Sum, - ); - } - let query_end = time_ranges as u64 * 1_000; - - group.bench_function(format!("{}/query_hot_cold", kind.slug()), |b| { - let store = store.clone(); - let mut query_idx = 0u64; - b.iter(|| { - let agg_id = if query_idx % 5 < 4 { - (query_idx % 2) + 1 - } else { - (query_idx % 8) + 3 - }; - query_idx += 1; - black_box( - store - .query_precomputed_output("test_metric", agg_id, 0, query_end) - .unwrap(), - ); - }); - }); - - group.bench_function(format!("{}/concurrent_hot_cold", kind.slug()), |b| { - let store = store.clone(); - b.iter(|| { - let barrier = Arc::new(Barrier::new(4)); - std::thread::scope(|scope| { - for t in 0..4usize { - let store = store.clone(); - let barrier = barrier.clone(); - scope.spawn(move || { - barrier.wait(); - for q in 0..50usize { - let idx = (t * 50 + q) as u64; - let agg_id = if idx % 5 < 4 { - (idx % 2) + 1 - } else { - (idx % 8) + 3 - }; - black_box( - store - .query_precomputed_output( - "test_metric", - agg_id, - 0, - query_end, - ) - .unwrap(), - ); - } - }); - } - }); - }); - }); - } - - group.finish(); -} - -criterion_group!( - benches, - bench_insert_batch_size, - bench_insert_num_agg_ids, - bench_query_range_store_size, - bench_query_exact_store_size, - bench_store_analyze, - bench_concurrent_reads, - bench_concurrent_writes, - bench_concurrent_mixed_read_write, - bench_cleanup_overhead, - bench_query_patterns, - bench_high_label_cardinality, - bench_multi_agg_id, -); -criterion_main!(benches); diff --git a/asap-query-engine/src/stores/simple_map_store/legacy/global.rs b/asap-query-engine/src/stores/simple_map_store/legacy/global.rs deleted file mode 100644 index 73f36a1..0000000 --- a/asap-query-engine/src/stores/simple_map_store/legacy/global.rs +++ /dev/null @@ -1,550 +0,0 @@ -use crate::data_model::{ - AggregateCore, AggregationType, CleanupPolicy, KeyByLabelValues, PrecomputedOutput, - StreamingConfig, -}; -use crate::stores::{Store, StoreResult, TimestampedBucketsMap}; -use std::collections::HashMap; -use std::sync::Arc; -use std::sync::Mutex; -use std::time::Instant; -use tracing::{debug, error, info}; - -type TimestampRange = (u64, u64); // (start_timestamp, end_timestamp) -type StoreKey = u64; // aggregation_id -type StoreValue = Vec<(Option, Box)>; - -/// In-memory storage implementation using single mutex (like Python version) -pub struct LegacySimpleMapStoreGlobal { - // Single global mutex protecting all data structures - lock: Mutex, - - // Store the streaming configuration - streaming_config: Arc, - - // Policy for cleaning up old aggregates - cleanup_policy: CleanupPolicy, -} - -struct StoreData { - // Main storage: aggregation_id -> (start_time, end_time) -> [(key, precompute)] - store: HashMap>, - - // Track metrics that have been created - metrics: std::collections::HashSet, - - // Count items inserted per metric for logging - items_inserted: HashMap, - - // Track earliest timestamp per aggregation ID - earliest_timestamp_per_aggregation_id: HashMap, - - // Track how many times each aggregate window has been read - read_counts: HashMap>, -} - -impl LegacySimpleMapStoreGlobal { - pub fn new(streaming_config: Arc, cleanup_policy: CleanupPolicy) -> Self { - Self { - lock: Mutex::new(StoreData { - store: HashMap::new(), - metrics: std::collections::HashSet::new(), - items_inserted: HashMap::new(), - earliest_timestamp_per_aggregation_id: HashMap::new(), - read_counts: HashMap::new(), - }), - streaming_config, - cleanup_policy, - } - } - - fn create_table(&self, data: &mut StoreData, metric: &str) { - // In the in-memory implementation, "creating a table" just means - // marking the metric as known - data.metrics.insert(metric.to_string()); - } - - fn cleanup_old_aggregates_fixed_count( - &self, - data: &mut StoreData, - metric: &str, - aggregation_id: u64, - num_aggregates_to_retain: Option, - ) { - // Return early if no retention limit configured - let configured_limit = match num_aggregates_to_retain { - Some(limit) => limit as usize, - None => return, - }; - - let retention_limit = configured_limit * 4; - let store_key = aggregation_id; - - // Get the time map for this store key - if let Some(time_map) = data.store.get_mut(&store_key) { - if time_map.len() <= retention_limit { - return; // Nothing to clean up - } - - // Collect all timestamp ranges and sort by start timestamp (oldest first) - let mut timestamp_windows: Vec = time_map.keys().copied().collect(); - timestamp_windows.sort_by_key(|&(start, _end)| start); - - // Calculate which ones to remove (oldest first) - let num_to_remove = timestamp_windows.len() - retention_limit; - let windows_to_remove: Vec = - timestamp_windows.into_iter().take(num_to_remove).collect(); - - // Remove old windows - for window in windows_to_remove { - if time_map.remove(&window).is_some() { - debug!( - "Removed old aggregate for {} aggregation_id {} window {}-{} (retention limit: {}, configured: {})", - metric, - aggregation_id, - window.0, - window.1, - retention_limit, - configured_limit - ); - } - } - } - } - - fn cleanup_old_aggregates_read_based( - &self, - data: &mut StoreData, - metric: &str, - aggregation_id: u64, - read_count_threshold: Option, - ) { - // Return early if no threshold configured - let threshold = match read_count_threshold { - Some(t) => t, - None => return, - }; - - let store_key = aggregation_id; - - // Get both the time map and read count map - let time_map = match data.store.get_mut(&store_key) { - Some(map) => map, - None => return, - }; - - let read_count_map = data.read_counts.entry(store_key).or_default(); - - // Collect windows where read_count >= threshold - let mut windows_to_remove: Vec = Vec::new(); - - for timestamp_range in time_map.keys() { - let read_count = read_count_map.get(timestamp_range).copied().unwrap_or(0); - - if read_count >= threshold { - windows_to_remove.push(*timestamp_range); - } - } - - // Remove windows that exceeded threshold - for window in &windows_to_remove { - if time_map.remove(window).is_some() { - let read_count = read_count_map.get(window).copied().unwrap_or(0); - read_count_map.remove(window); - - debug!( - "Removed aggregate for {} aggregation_id {} window {}-{} (read_count: {} >= threshold: {})", - metric, - aggregation_id, - window.0, - window.1, - read_count, - threshold - ); - } - } - } - - fn cleanup_old_aggregates( - &self, - data: &mut StoreData, - metric: &str, - aggregation_id: u64, - num_aggregates_to_retain: Option, - read_count_threshold: Option, - ) { - match self.cleanup_policy { - CleanupPolicy::CircularBuffer => { - self.cleanup_old_aggregates_fixed_count( - data, - metric, - aggregation_id, - num_aggregates_to_retain, - ); - } - CleanupPolicy::ReadBased => { - self.cleanup_old_aggregates_read_based( - data, - metric, - aggregation_id, - read_count_threshold, - ); - } - CleanupPolicy::NoCleanup => { - // Do nothing - no cleanup - } - } - } -} - -#[async_trait::async_trait] -impl Store for LegacySimpleMapStoreGlobal { - fn insert_precomputed_output( - &self, - output: PrecomputedOutput, - precompute: Box, - ) -> StoreResult<()> { - self.insert_precomputed_output_batch(vec![(output, precompute)]) - } - - fn insert_precomputed_output_batch( - &self, - outputs: Vec<(PrecomputedOutput, Box)>, - ) -> StoreResult<()> { - let batch_insert_start_time = Instant::now(); - let batch_size = outputs.len(); - - // Measure lock acquisition time - #[cfg(feature = "lock_profiling")] - let lock_wait_start = Instant::now(); - - // Single lock for entire batch (like Python version) - let mut data = self.lock.lock().unwrap(); - - #[cfg(feature = "lock_profiling")] - { - let lock_wait_duration = lock_wait_start.elapsed(); - info!( - "🔒 Insert lock wait time: {:.2}ms (batch_size: {})", - lock_wait_duration.as_secs_f64() * 1000.0, - batch_size - ); - } - - #[cfg(feature = "lock_profiling")] - let lock_hold_start = Instant::now(); - - for (output, precompute) in outputs { - let aggregation_config = self - .streaming_config - .get_aggregation_config(output.aggregation_id); - - if aggregation_config.is_none() { - error!( - "Aggregation config not found for aggregation_id {}. Skipping insert.", - output.aggregation_id - ); - continue; - } - let aggregation_config = aggregation_config.unwrap(); - - let metric = aggregation_config.metric.clone(); - let aggregation_id = output.aggregation_id; - - // Create table if it doesn't exist - if !data.metrics.contains(&metric) { - self.create_table(&mut data, &metric); - } - - // Update earliest timestamp tracking - if let Some(current_earliest) = data - .earliest_timestamp_per_aggregation_id - .get_mut(&aggregation_id) - { - if output.start_timestamp < *current_earliest { - *current_earliest = output.start_timestamp; - } - } else { - data.earliest_timestamp_per_aggregation_id - .insert(aggregation_id, output.start_timestamp); - } - - let store_key = aggregation_id; - let timestamp_range = (output.start_timestamp, output.end_timestamp); - - // Get or create the time-based map for this aggregation - let time_map = data.store.entry(store_key).or_default(); - - // Get or create the value vector for this timestamp range - let store_value = time_map.entry(timestamp_range).or_default(); - - // Add the new entry with the real precompute data - store_value.push((output.key, precompute)); - - // Apply retention policy if configured (but exclude DeltaSetAggregator) - if aggregation_config.aggregation_type != AggregationType::DeltaSetAggregator { - self.cleanup_old_aggregates( - &mut data, - &metric, - aggregation_id, - aggregation_config.num_aggregates_to_retain, - aggregation_config.read_count_threshold, - ); - } - - // Update insertion count - let current_count = data.items_inserted.entry(metric.clone()).or_insert(0); - *current_count += 1; - - if (*current_count).is_multiple_of(1000) { - debug!("Inserted {} items into {}", current_count, metric); - } - } - - #[cfg(feature = "lock_profiling")] - { - let lock_hold_duration = lock_hold_start.elapsed(); - info!( - "🔓 Insert lock hold time: {:.2}ms (batch_size: {})", - lock_hold_duration.as_secs_f64() * 1000.0, - batch_size - ); - } - - // Lock will be dropped here when `data` goes out of scope - - let batch_insert_duration = batch_insert_start_time.elapsed(); - debug!( - "Batch insert of {} items took: {:.2}ms", - batch_size, - batch_insert_duration.as_secs_f64() * 1000.0 - ); - Ok(()) - } - - fn query_precomputed_output( - &self, - metric: &str, - aggregation_id: u64, - start: u64, - end: u64, - ) -> Result> { - let query_start_time = Instant::now(); - let store_key = aggregation_id; - - // Measure lock acquisition time - #[cfg(feature = "lock_profiling")] - let lock_wait_start = Instant::now(); - - // Single lock for entire query - now mutable to track read counts - let mut data = self.lock.lock().unwrap(); - - #[cfg(feature = "lock_profiling")] - { - let lock_wait_duration = lock_wait_start.elapsed(); - info!( - "🔒 Query lock wait time: {:.2}ms (metric: {}, agg_id: {})", - lock_wait_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id - ); - } - - #[cfg(feature = "lock_profiling")] - let lock_hold_start = Instant::now(); - - let time_map = match data.store.get(&store_key) { - Some(map) => map, - None => { - info!("Metric {} not found in store", metric); - return Ok(HashMap::new()); - } - }; - - let mut results: TimestampedBucketsMap = HashMap::new(); - let mut total_entries = 0; - - // Find all timestamp ranges that overlap with our query range - let range_scan_start_time = Instant::now(); - - // First, collect all matching timestamp ranges - let mut matching_ranges: Vec = time_map - .keys() - .filter(|(range_start, range_end)| start <= *range_start && end >= *range_end) - .copied() - .collect(); - - // Sort by start timestamp to ensure chronological order - // This is important for range queries that use sliding windows - matching_ranges.sort_by_key(|(range_start, _)| *range_start); - - // Now iterate in sorted order, including timestamp with each bucket - for timestamp_range in &matching_ranges { - if let Some(store_values) = time_map.get(timestamp_range) { - for (key_opt, precompute) in store_values.iter() { - results - .entry(key_opt.clone()) - .or_default() - .push((*timestamp_range, precompute.clone_boxed_core().into())); - - total_entries += 1; - } - } - } - - // Update read counts for accessed ranges (after we're done with time_map to avoid borrow conflicts) - let read_count_map = data.read_counts.entry(store_key).or_default(); - for timestamp_range in &matching_ranges { - *read_count_map.entry(*timestamp_range).or_insert(0) += 1; - } - - let range_scan_duration = range_scan_start_time.elapsed(); - debug!( - "Range scanning took: {:.2}ms", - range_scan_duration.as_secs_f64() * 1000.0 - ); - - let query_duration = query_start_time.elapsed(); - debug!( - "Total query took: {:.2}ms", - query_duration.as_secs_f64() * 1000.0 - ); - - debug!( - "Found {} entries for query on {} (aggregation_id: {}, start: {}, end: {})", - total_entries, metric, aggregation_id, start, end - ); - debug!("Found {} unique keys", results.len()); - - #[cfg(feature = "lock_profiling")] - { - let lock_hold_duration = lock_hold_start.elapsed(); - info!( - "🔓 Query lock hold time: {:.2}ms (metric: {}, agg_id: {}, entries: {})", - lock_hold_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id, - total_entries - ); - } - - // Lock will be dropped here when `data` goes out of scope - - Ok(results) - } - - fn query_precomputed_output_exact( - &self, - metric: &str, - aggregation_id: u64, - exact_start: u64, - exact_end: u64, - ) -> Result> { - let query_start_time = Instant::now(); - let store_key = aggregation_id; - - // Measure lock acquisition time - #[cfg(feature = "lock_profiling")] - let lock_wait_start = Instant::now(); - - let mut data = self.lock.lock().unwrap(); - - #[cfg(feature = "lock_profiling")] - { - let lock_wait_duration = lock_wait_start.elapsed(); - info!( - "🔒 Exact query lock wait time: {:.2}ms (metric: {}, agg_id: {})", - lock_wait_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id - ); - } - - #[cfg(feature = "lock_profiling")] - let lock_hold_start = Instant::now(); - - let time_map = match data.store.get(&store_key) { - Some(map) => map, - None => { - debug!("Metric {} not found in store for exact query", metric); - return Ok(HashMap::new()); - } - }; - - let mut results: TimestampedBucketsMap = HashMap::new(); - - // Look for exact timestamp match (strict - no tolerance) - let timestamp_range = (exact_start, exact_end); - let mut found_match = false; - - // First, collect the results (immutable borrow of time_map) - if let Some(store_values) = time_map.get(×tamp_range) { - found_match = true; - - // Collect results with timestamp - let mut total_entries = 0; - for (key_opt, precompute) in store_values.iter() { - results - .entry(key_opt.clone()) - .or_default() - .push((timestamp_range, precompute.clone_boxed_core().into())); - total_entries += 1; - } - - debug!( - "Exact match FOUND for [{}, {}]: {} entries across {} keys", - exact_start, - exact_end, - total_entries, - results.len() - ); - } else { - debug!( - "Exact match NOT FOUND for metric: {}, agg_id: {}, range: [{}, {}]", - metric, aggregation_id, exact_start, exact_end - ); - } - - // Now update read count (mutable borrow of data.read_counts) - // This happens after we're done with time_map - if found_match { - let read_count_map = data.read_counts.entry(store_key).or_default(); - *read_count_map.entry(timestamp_range).or_insert(0) += 1; - } - - #[cfg(feature = "lock_profiling")] - { - let lock_hold_duration = lock_hold_start.elapsed(); - info!( - "🔓 Exact query lock hold time: {:.2}ms (metric: {}, agg_id: {}, found: {})", - lock_hold_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id, - !results.is_empty() - ); - } - - let query_duration = query_start_time.elapsed(); - debug!( - "Exact timestamp query took: {:.2}ms (found: {})", - query_duration.as_secs_f64() * 1000.0, - !results.is_empty() - ); - - // Lock will be dropped here when `data` goes out of scope - - Ok(results) - } - - fn get_earliest_timestamp_per_aggregation_id( - &self, - ) -> Result, Box> { - let data = self.lock.lock().unwrap(); - Ok(data.earliest_timestamp_per_aggregation_id.clone()) - } - - fn close(&self) -> StoreResult<()> { - // For in-memory store, no cleanup needed - info!("LegacySimpleMapStoreGlobal closed"); - Ok(()) - } -} diff --git a/asap-query-engine/src/stores/simple_map_store/legacy/mod.rs b/asap-query-engine/src/stores/simple_map_store/legacy/mod.rs deleted file mode 100644 index 24a12f4..0000000 --- a/asap-query-engine/src/stores/simple_map_store/legacy/mod.rs +++ /dev/null @@ -1,5 +0,0 @@ -mod global; -mod per_key; - -pub use global::LegacySimpleMapStoreGlobal; -pub use per_key::LegacySimpleMapStorePerKey; diff --git a/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs b/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs deleted file mode 100644 index 58cd41f..0000000 --- a/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs +++ /dev/null @@ -1,639 +0,0 @@ -use crate::data_model::{ - AggregateCore, AggregationType, CleanupPolicy, KeyByLabelValues, PrecomputedOutput, - StreamingConfig, -}; -use crate::stores::{Store, StoreResult, TimestampedBucketsMap}; -use dashmap::DashMap; -use std::collections::HashMap; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, RwLock}; -use std::time::Instant; -use tracing::{debug, error, info}; - -type TimestampRange = (u64, u64); // (start_timestamp, end_timestamp) -type StoreKey = u64; // aggregation_id -type StoreValue = Vec<(Option, Box)>; - -/// Per-aggregation_id data protected by RwLock -struct StoreKeyData { - // Main storage: (start_time, end_time) -> [(key, precompute)] - time_map: HashMap, - - // Track how many times each timestamp range has been read - read_counts: HashMap, -} - -impl StoreKeyData { - fn new() -> Self { - Self { - time_map: HashMap::new(), - read_counts: HashMap::new(), - } - } -} - -/// In-memory storage implementation using per-key locks for concurrency -pub struct LegacySimpleMapStorePerKey { - // Lock-free concurrent outer map - per aggregation_id - store: DashMap>>, - - // Separate concurrent maps for global state - earliest_timestamps: DashMap, - metrics: DashMap, // HashSet equivalent - items_inserted: DashMap, - - // Store the streaming configuration - streaming_config: Arc, - - // Policy for cleaning up old aggregates - cleanup_policy: CleanupPolicy, -} - -impl LegacySimpleMapStorePerKey { - pub fn new(streaming_config: Arc, cleanup_policy: CleanupPolicy) -> Self { - Self { - store: DashMap::new(), - earliest_timestamps: DashMap::new(), - metrics: DashMap::new(), - items_inserted: DashMap::new(), - streaming_config, - cleanup_policy, - } - } - - fn cleanup_old_aggregates_fixed_count( - &self, - data: &mut StoreKeyData, - metric: &str, - aggregation_id: u64, - num_aggregates_to_retain: Option, - ) { - // Return early if no retention limit configured - let configured_limit = match num_aggregates_to_retain { - Some(limit) => limit as usize, - None => return, - }; - - let retention_limit = configured_limit * 4; - - if data.time_map.len() <= retention_limit { - return; // Nothing to clean up - } - - // Collect all timestamp ranges and sort by start timestamp (oldest first) - let mut timestamp_windows: Vec = data.time_map.keys().copied().collect(); - timestamp_windows.sort_by_key(|&(start, _end)| start); - - // Calculate which ones to remove (oldest first) - let num_to_remove = timestamp_windows.len() - retention_limit; - let windows_to_remove: Vec = - timestamp_windows.into_iter().take(num_to_remove).collect(); - - // Remove old windows from both time_map and read_counts - for window in windows_to_remove { - if data.time_map.remove(&window).is_some() { - data.read_counts.remove(&window); // Also remove from read_counts - debug!( - "Removed old aggregate for {} aggregation_id {} window {}-{} (retention limit: {}, configured: {})", - metric, - aggregation_id, - window.0, - window.1, - retention_limit, - configured_limit - ); - } - } - } - - fn cleanup_old_aggregates_read_based( - &self, - data: &mut StoreKeyData, - metric: &str, - aggregation_id: u64, - read_count_threshold: Option, - ) { - // Return early if no threshold configured - let threshold = match read_count_threshold { - Some(t) => t, - None => return, - }; - - // Collect windows where read_count >= threshold - let mut windows_to_remove: Vec = Vec::new(); - - for timestamp_range in data.time_map.keys() { - let read_count = data.read_counts.get(timestamp_range).copied().unwrap_or(0); - - if read_count >= threshold { - windows_to_remove.push(*timestamp_range); - } - } - - // Remove windows that exceeded threshold - for window in &windows_to_remove { - //if let Some(_) = data.time_map.remove(window) { - if data.time_map.remove(window).is_some() { - let read_count = data.read_counts.get(window).copied().unwrap_or(0); - data.read_counts.remove(window); - - debug!( - "Removed aggregate for {} aggregation_id {} window {}-{} (read_count: {} >= threshold: {})", - metric, - aggregation_id, - window.0, - window.1, - read_count, - threshold - ); - } - } - } - - fn cleanup_old_aggregates( - &self, - data: &mut StoreKeyData, - metric: &str, - aggregation_id: u64, - num_aggregates_to_retain: Option, - read_count_threshold: Option, - ) { - match self.cleanup_policy { - CleanupPolicy::CircularBuffer => { - self.cleanup_old_aggregates_fixed_count( - data, - metric, - aggregation_id, - num_aggregates_to_retain, - ); - } - CleanupPolicy::ReadBased => { - self.cleanup_old_aggregates_read_based( - data, - metric, - aggregation_id, - read_count_threshold, - ); - } - CleanupPolicy::NoCleanup => { - // Do nothing - no cleanup - } - } - } - - fn insert_for_store_key( - &self, - store_key: &StoreKey, - metric: &str, - items: Vec<(PrecomputedOutput, Box)>, - ) -> StoreResult<()> { - let aggregation_id = *store_key; - - // Measure lock acquisition time - #[cfg(feature = "lock_profiling")] - let lock_wait_start = Instant::now(); - - // Get or create the store data for this key - let store_data_lock = self - .store - .entry(*store_key) - .or_insert_with(|| Arc::new(RwLock::new(StoreKeyData::new()))); - - #[cfg(feature = "lock_profiling")] - { - let lock_wait_duration = lock_wait_start.elapsed(); - info!( - "🔒 Insert DashMap get time: {:.2}ms (metric: {}, agg_id: {}, items: {})", - lock_wait_duration.as_secs_f64() * 1000.0, - metric, - *store_key, - items.len() - ); - } - - #[cfg(feature = "lock_profiling")] - let rwlock_wait_start = Instant::now(); - - // Acquire write lock for this aggregation_id only - let mut data = store_data_lock.write().map_err(|e| { - format!( - "Failed to acquire write lock for aggregation_id {}: {}", - store_key, e - ) - })?; - - #[cfg(feature = "lock_profiling")] - { - let rwlock_wait_duration = rwlock_wait_start.elapsed(); - info!( - "🔒 Insert RwLock wait time: {:.2}ms (metric: {}, agg_id: {}, items: {})", - rwlock_wait_duration.as_secs_f64() * 1000.0, - metric, - *store_key, - items.len() - ); - } - - #[cfg(feature = "lock_profiling")] - let lock_hold_start = Instant::now(); - - for (output, precompute) in items { - // Create metric if needed (lock-free DashMap insert) - self.metrics.entry(metric.to_string()).or_insert(()); - - // Update earliest timestamp (lock-free atomic operation) - self.earliest_timestamps - .entry(aggregation_id) - .and_modify(|earliest| { - let current = earliest.load(Ordering::Relaxed); - if output.start_timestamp < current { - earliest.store(output.start_timestamp, Ordering::Relaxed); - } - }) - .or_insert_with(|| AtomicU64::new(output.start_timestamp)); - - // Insert into time map - let timestamp_range = (output.start_timestamp, output.end_timestamp); - data.time_map - .entry(timestamp_range) - .or_default() - .push((output.key, precompute)); - - // Update insertion count (lock-free atomic increment) - self.items_inserted - .entry(metric.to_string()) - .and_modify(|count| { - let new_count = count.fetch_add(1, Ordering::Relaxed) + 1; - if new_count.is_multiple_of(1000) { - debug!("Inserted {} items into {}", new_count, metric); - } - }) - .or_insert_with(|| AtomicU64::new(1)); - } - - // Apply retention policy if configured (but exclude DeltaSetAggregator) - let aggregation_config = self - .streaming_config - .get_aggregation_config(aggregation_id) - .ok_or_else(|| format!("Aggregation config not found for {}", aggregation_id))?; - - if aggregation_config.aggregation_type != AggregationType::DeltaSetAggregator { - self.cleanup_old_aggregates( - &mut data, - metric, - aggregation_id, - aggregation_config.num_aggregates_to_retain, - aggregation_config.read_count_threshold, - ); - } - - #[cfg(feature = "lock_profiling")] - { - let lock_hold_duration = lock_hold_start.elapsed(); - info!( - "🔓 Insert lock hold time: {:.2}ms (metric: {}, agg_id: {})", - lock_hold_duration.as_secs_f64() * 1000.0, - metric, - *store_key - ); - } - - Ok(()) - } -} - -#[async_trait::async_trait] -impl Store for LegacySimpleMapStorePerKey { - fn insert_precomputed_output( - &self, - output: PrecomputedOutput, - precompute: Box, - ) -> StoreResult<()> { - self.insert_precomputed_output_batch(vec![(output, precompute)]) - } - - fn insert_precomputed_output_batch( - &self, - outputs: Vec<(PrecomputedOutput, Box)>, - ) -> StoreResult<()> { - let batch_insert_start_time = Instant::now(); - let batch_size = outputs.len(); - - // Group by aggregation_id - #[allow(clippy::type_complexity)] - let mut grouped: HashMap< - StoreKey, - (String, Vec<(PrecomputedOutput, Box)>), - > = HashMap::new(); - - for (output, precompute) in outputs { - let aggregation_config = self - .streaming_config - .get_aggregation_config(output.aggregation_id); - - if aggregation_config.is_none() { - error!( - "Aggregation config not found for aggregation_id {}. Skipping insert.", - output.aggregation_id - ); - continue; - } - let aggregation_config = aggregation_config.unwrap(); - - let metric = aggregation_config.metric.clone(); - let store_key = output.aggregation_id; - - grouped - .entry(store_key) - .or_insert_with(|| (metric.clone(), Vec::new())) - .1 - .push((output, precompute)); - } - - // Sort keys to avoid deadlock when acquiring multiple locks - let mut keys: Vec<_> = grouped.keys().cloned().collect(); - keys.sort(); - - // Process each group - for store_key in keys { - let (metric, items) = grouped.remove(&store_key).unwrap(); - self.insert_for_store_key(&store_key, &metric, items)?; - } - - let batch_insert_duration = batch_insert_start_time.elapsed(); - debug!( - "Batch insert of {} items took: {:.2}ms", - batch_size, - batch_insert_duration.as_secs_f64() * 1000.0 - ); - Ok(()) - } - - fn query_precomputed_output( - &self, - metric: &str, - aggregation_id: u64, - start: u64, - end: u64, - ) -> Result> { - let query_start_time = Instant::now(); - let store_key = aggregation_id; - - // Measure lock acquisition time - #[cfg(feature = "lock_profiling")] - let lock_wait_start = Instant::now(); - - // Get the store data for this aggregation_id - let store_data_lock = match self.store.get(&store_key) { - Some(lock) => lock, - None => { - info!("Metric {} not found in store", metric); - return Ok(HashMap::new()); - } - }; - - #[cfg(feature = "lock_profiling")] - { - let lock_wait_duration = lock_wait_start.elapsed(); - info!( - "🔒 Query DashMap get time: {:.2}ms (metric: {}, agg_id: {})", - lock_wait_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id - ); - } - - #[cfg(feature = "lock_profiling")] - let rwlock_wait_start = Instant::now(); - - // Acquire write lock (needed to update read_counts) - let mut data = store_data_lock.write().map_err(|e| { - format!( - "Failed to acquire write lock for query aggregation_id {}: {}", - store_key, e - ) - })?; - - #[cfg(feature = "lock_profiling")] - { - let rwlock_wait_duration = rwlock_wait_start.elapsed(); - info!( - "🔒 Query RwLock wait time: {:.2}ms (metric: {}, agg_id: {})", - rwlock_wait_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id - ); - } - - #[cfg(feature = "lock_profiling")] - let lock_hold_start = Instant::now(); - - let mut results: TimestampedBucketsMap = HashMap::new(); - let mut total_entries = 0; - - // Find all timestamp ranges that overlap with our query range - let range_scan_start_time = Instant::now(); - - // First, collect all matching timestamp ranges - let mut matching_ranges: Vec = data - .time_map - .keys() - .filter(|(range_start, range_end)| start <= *range_start && end >= *range_end) - .copied() - .collect(); - - // Sort by start timestamp to ensure chronological order - // This is important for range queries that use sliding windows - matching_ranges.sort_by_key(|(range_start, _)| *range_start); - - // Now iterate in sorted order, including timestamp with each bucket - for timestamp_range in &matching_ranges { - if let Some(store_values) = data.time_map.get(timestamp_range) { - for (key_opt, precompute) in store_values.iter() { - results - .entry(key_opt.clone()) - .or_default() - .push((*timestamp_range, precompute.clone_boxed_core().into())); - - total_entries += 1; - } - } - } - - // Update read counts for accessed ranges - for timestamp_range in &matching_ranges { - *data.read_counts.entry(*timestamp_range).or_insert(0) += 1; - } - - let range_scan_duration = range_scan_start_time.elapsed(); - debug!( - "Range scanning took: {:.2}ms", - range_scan_duration.as_secs_f64() * 1000.0 - ); - - let query_duration = query_start_time.elapsed(); - debug!( - "Total query took: {:.2}ms", - query_duration.as_secs_f64() * 1000.0 - ); - - debug!( - "Found {} entries for query on {} (aggregation_id: {}, start: {}, end: {})", - total_entries, metric, aggregation_id, start, end - ); - debug!("Found {} unique keys", results.len()); - - #[cfg(feature = "lock_profiling")] - { - let lock_hold_duration = lock_hold_start.elapsed(); - info!( - "🔓 Query lock hold time: {:.2}ms (metric: {}, agg_id: {}, entries: {})", - lock_hold_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id, - total_entries - ); - } - - Ok(results) - } - - fn query_precomputed_output_exact( - &self, - metric: &str, - aggregation_id: u64, - exact_start: u64, - exact_end: u64, - ) -> Result> { - let query_start_time = Instant::now(); - let store_key = aggregation_id; - - // Measure lock acquisition time - #[cfg(feature = "lock_profiling")] - let lock_wait_start = Instant::now(); - - // Get the store data for this aggregation_id - let store_data_lock = match self.store.get(&store_key) { - Some(lock) => lock, - None => { - debug!("Metric {} not found in store for exact query", metric); - return Ok(HashMap::new()); - } - }; - - #[cfg(feature = "lock_profiling")] - { - let lock_wait_duration = lock_wait_start.elapsed(); - info!( - "🔒 Exact query DashMap get time: {:.2}ms (metric: {}, agg_id: {})", - lock_wait_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id - ); - } - - #[cfg(feature = "lock_profiling")] - let rwlock_wait_start = Instant::now(); - - // Acquire write lock (needed to update read_counts) - let mut data = store_data_lock.write().map_err(|e| { - format!( - "Failed to acquire write lock for exact query aggregation_id {}: {}", - store_key, e - ) - })?; - - #[cfg(feature = "lock_profiling")] - { - let rwlock_wait_duration = rwlock_wait_start.elapsed(); - info!( - "🔒 Exact query RwLock wait time: {:.2}ms (metric: {}, agg_id: {})", - rwlock_wait_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id - ); - } - - #[cfg(feature = "lock_profiling")] - let lock_hold_start = Instant::now(); - - let mut results: TimestampedBucketsMap = HashMap::new(); - - // Look for exact timestamp match (strict - no tolerance) - let timestamp_range = (exact_start, exact_end); - let mut found_match = false; - - // First, collect the results (immutable borrow of time_map) - if let Some(store_values) = data.time_map.get(×tamp_range) { - found_match = true; - - // Collect results with timestamp - let mut total_entries = 0; - for (key_opt, precompute) in store_values.iter() { - results - .entry(key_opt.clone()) - .or_default() - .push((timestamp_range, precompute.clone_boxed_core().into())); - total_entries += 1; - } - - debug!( - "Exact match FOUND for [{}, {}]: {} entries across {} keys", - exact_start, - exact_end, - total_entries, - results.len() - ); - } else { - debug!( - "Exact match NOT FOUND for metric: {}, agg_id: {}, range: [{}, {}]", - metric, aggregation_id, exact_start, exact_end - ); - } - - // Now update read count (mutable borrow of data.read_counts) - if found_match { - *data.read_counts.entry(timestamp_range).or_insert(0) += 1; - } - - #[cfg(feature = "lock_profiling")] - { - let lock_hold_duration = lock_hold_start.elapsed(); - info!( - "🔓 Exact query lock hold time: {:.2}ms (metric: {}, agg_id: {}, found: {})", - lock_hold_duration.as_secs_f64() * 1000.0, - metric, - aggregation_id, - !results.is_empty() - ); - } - - let query_duration = query_start_time.elapsed(); - debug!( - "Exact timestamp query took: {:.2}ms (found: {})", - query_duration.as_secs_f64() * 1000.0, - !results.is_empty() - ); - - Ok(results) - } - - fn get_earliest_timestamp_per_aggregation_id( - &self, - ) -> Result, Box> { - // No lock needed - DashMap with AtomicU64 - let result = self - .earliest_timestamps - .iter() - .map(|entry| (*entry.key(), entry.value().load(Ordering::Relaxed))) - .collect(); - - Ok(result) - } - - fn close(&self) -> StoreResult<()> { - // For in-memory store, no cleanup needed - info!("LegacySimpleMapStorePerKey closed"); - Ok(()) - } -} diff --git a/asap-query-engine/src/stores/simple_map_store/mod.rs b/asap-query-engine/src/stores/simple_map_store/mod.rs index 24773bd..a8a6d32 100644 --- a/asap-query-engine/src/stores/simple_map_store/mod.rs +++ b/asap-query-engine/src/stores/simple_map_store/mod.rs @@ -1,7 +1,5 @@ mod common; pub mod global; -#[cfg(feature = "legacy_stores")] -pub mod legacy; pub mod per_key; use crate::data_model::{