update

Author: zzylol

asap-query-engine/src/stores/simple_map_store/INDEX_DESIGN.md

@@ -2,137 +2,200 @@
 
 ## Overview
 
-The `SimpleMapStore` uses an **inverted index** (label-primary) layout to store precomputed aggregates. This design aligns the storage structure with the query return type (`HashMap<Option<KeyByLabelValues>, Vec<TimestampedBucket>>`), eliminating the need for regrouping at query time.
+`SimpleMapStore` uses an **epoch-partitioned inverted index** to store precomputed aggregates. Three VictoriaMetrics-inspired optimizations are applied on top of the basic label-primary layout:
 
-## Data Structure
+1. **Label Interning** — label combinations are mapped to compact `MetricID` (u32), reducing key size and hash cost.
+2. **Epoch Partitioning** — data is split into fixed-capacity epoch slots; the oldest epoch is dropped O(1) when the cap is exceeded (CircularBuffer policy).
+3. **Sorted Vec Posting Lists** — the reverse index (`window_to_ids`) stores `Vec<MetricID>` maintained in sorted order, enabling binary-search deduplication on insert and cache-friendly iteration on lookup.
 
-### Per-Key Store (`per_key.rs`)
+---
 
-Each `aggregation_id` maps to a `StoreKeyData` protected by an `RwLock`:
+## Data Structures
+
+### Types (common.rs)
+
+```rust
+pub type MetricID = u32;          // compact interned label ID
+pub type EpochID  = u64;          // monotonically increasing epoch counter
+pub type TimestampRange = (u64, u64);  // (start_timestamp, end_timestamp)
+pub type MetricBucketMap = HashMap<MetricID, Vec<(TimestampRange, Arc<dyn AggregateCore>)>>;
+```
+
+### InternTable (common.rs)
+
+```
+InternTable {
+    label_to_id: HashMap<Option<KeyByLabelValues>, MetricID>
+    id_to_label: Vec<Option<KeyByLabelValues>>
+}
+```
+
+- `intern(label)` → O(1) amortized, no double-hashing (uses `HashMap::entry`)
+- `resolve(id)` → O(1) indexed Vec lookup
+- All internal index maps use `MetricID` (u32) as keys, not full label strings
+
+### EpochData (common.rs)
+
+One epoch holds up to `epoch_capacity` distinct time windows.
+
+```
+EpochData {
+    label_map:     HashMap<MetricID, BTreeMap<TimestampRange, Vec<Arc<dyn AggregateCore>>>>
+    window_to_ids: HashMap<TimestampRange, Vec<MetricID>>   // sorted (Optimization 3)
+    time_ranges:   BTreeSet<TimestampRange>
+}
+```
+
+- **`label_map`** (primary index): inverted index MetricID → time-sorted BTreeMap of aggregates. Enables O(log N + k) range queries per label.
+- **`window_to_ids`** (reverse index): for each time window, sorted `Vec<MetricID>` of labels that contain data. Used for exact queries and targeted cleanup without full label scans.
+- **`time_ranges`** (secondary index): all distinct windows in this epoch, sorted. Used for epoch range filtering (skip epochs that don't overlap the query interval) and cleanup ordering.
+
+### Per-Key Store (per_key.rs)
+
+Each aggregation_id gets its own `StoreKeyData` behind a per-key `RwLock`:
 
 ```
 DashMap<aggregation_id, Arc<RwLock<StoreKeyData>>>
 
 StoreKeyData {
-    label_map:         HashMap<Option<KeyByLabelValues>, BTreeMap<(start, end), Vec<Arc<dyn AggregateCore>>>>
-    window_to_labels:  HashMap<(start, end), HashSet<Option<KeyByLabelValues>>>
-    time_ranges: BTreeSet<(start, end)>
-    read_counts: Mutex<HashMap<(start, end), u64>>
+    intern:           InternTable
+    epochs:           BTreeMap<EpochID, EpochData>
+    current_epoch_id: EpochID
+    epoch_capacity:   Option<usize>   // None = unlimited
+    max_epochs:       usize           // default 4
+    read_counts:      Mutex<HashMap<TimestampRange, u64>>
 }
 ```
 
-- **`label_map`** (primary index): Inverted index from label key to a time-sorted BTreeMap of aggregates. Enables O(log n + k) range queries per label.
-- **`window_to_labels`** (reverse index): For each time window, tracks exactly which labels contain data. Enables exact queries and cleanup to avoid full label scans.
-- **`time_ranges`** (secondary index): All known timestamp ranges across all labels. Used for cleanup counting and read-count tracking.
-- **`read_counts`**: Wrapped in `Mutex` so queries can use a read lock on the outer `RwLock` (only needs brief exclusive access to increment counts).
+`read_counts` is behind an inner `Mutex` so queries can hold a read lock on the outer `RwLock` and still update counts (brief inner lock, no write-lock upgrade needed).
 
-### Global Store (`global.rs`)
+### Global Store (global.rs)
 
-Same inverted index structure, but nested under a single `Mutex<StoreData>`:
+Same per-key epoch structure, but all aggregation_ids share a single `Mutex<StoreData>`:
 
 ```
 Mutex<StoreData>
 
 StoreData {
-    store:            HashMap<aggregation_id, HashMap<Option<KeyByLabelValues>, BTreeMap<(start, end), Vec<Arc<dyn AggregateCore>>>>>
-    window_to_labels: HashMap<aggregation_id, HashMap<(start, end), HashSet<Option<KeyByLabelValues>>>>
-    time_ranges:      HashMap<aggregation_id, BTreeSet<(start, end)>>
-    read_counts:      HashMap<aggregation_id, HashMap<(start, end), u64>>
+    stores:      HashMap<aggregation_id, PerKeyState>
+    read_counts: HashMap<aggregation_id, HashMap<TimestampRange, u64>>
+}
+
+PerKeyState {
+    intern:           InternTable
+    epochs:           BTreeMap<EpochID, EpochData>
+    current_epoch_id: EpochID
+    epoch_capacity:   Option<usize>
+    max_epochs:       usize
 }
 ```
 
-No inner Mutex for `read_counts` since the outer Mutex already serializes all access.
+No inner `Mutex` for `read_counts` — the outer `Mutex` already serializes all access.
+
+---
 
 ## Theoretical Complexity
 
 ### Variables
 
 | Symbol | Meaning |
-|---|---|
+|--------|---------|
 | A | Number of distinct aggregation IDs |
 | L | Number of distinct label combinations (cardinality) |
 | N | Number of distinct time windows stored per (agg_id, label) |
-| k | Number of results matched or entries removed in a given operation |
+| E | Number of epochs (bounded by `max_epochs`, default 4) |
+| k | Number of results matched or entries removed |
 | m | Number of labels present in a specific time window |
-| V | Number of aggregate objects stored per (label, window) slot (typically 1) |
+| V | Aggregate objects per (label, window) slot (typically 1) |
 
 ### Time Complexity
 
 | Operation | Time | Notes |
-|---|---|---|
-| **Insert** (single entry) | **O(log N)** | DashMap O(1) + RwLock O(1) + HashMap O(1) + BTreeMap O(log N) + BTreeSet O(log N) |
-| **Insert** (batch of B entries, same agg_id) | **O(B · log N)** | One write-lock acquisition amortized over B items |
-| **Range query** | **O(L · (log N + k))** | BTreeMap::range per label in O(log N + k_L); results already grouped by label |
-| **Exact query** | **O(m · log N)** | window_to_labels lookup O(1) + BTreeMap point get O(log N) per matching label |
-| **CircularBuffer cleanup** | **O(k · m)** amortized | BTreeSet iteration O(k) + targeted label-map removals via window_to_labels |
-| **ReadBased cleanup** | **O(N + k · m)** | Full read_counts scan O(N) + targeted removals O(k · m) |
-| **get_earliest_timestamp** | **O(A)** | DashMap iteration over A entries with atomic loads |
+|-----------|------|-------|
+| **Insert** (single entry) | O(log N) | DashMap O(1) + RwLock O(1) + InternTable O(1) + BTreeMap O(log N) + BTreeSet O(log N) + sorted-Vec insert O(L) worst |
+| **Insert** (batch B, same agg_id) | O(B · log N) | One write-lock acquisition amortized over B items |
+| **Epoch rotation** (CircularBuffer) | O(1) amortized | BTreeMap insert new epoch + BTreeMap pop oldest |
+| **Range query** | O(E · L · (log N + k)) | Per epoch: skip check O(1) + range scan per label O(log N + k_L); MetricID→label resolution O(L) |
+| **Exact query** | O(E · m · log N) | Per epoch: reverse-index lookup O(1) + point get O(log N) per matching label; stops at first match |
+| **CircularBuffer cleanup** | O(1) amortized | Epoch rotation drops entire oldest epoch |
+| **ReadBased cleanup** | O(N + k · m) | Scan read_counts O(N) + targeted removals via window_to_ids O(k · m) |
+| **get_earliest_timestamp** | O(A) | DashMap iteration with AtomicU64 loads |
 
 ### Space Complexity
 
 | Structure | Space | Notes |
-|---|---|---|
-| `label_map` | O(A · L · N · V) | Primary index: agg_id → label → BTreeMap(window → Vec<Arc<Agg>>) |
-| `window_to_labels` | O(A · N · L) | Reverse index: agg_id → window → HashSet\<label\> |
-| `time_ranges` | O(A · N) | Secondary index: agg_id → BTreeSet of all windows |
-| `read_counts` | O(A · N) | agg_id → HashMap\<window, u64\> |
-| **Total** | **O(A · L · N · V)** | Dominated by the primary label_map |
-
-Arc-sharing means query results hold references into the store; no deep copies are made for read paths.
-
-### Operation Complexity Summary
-
-| Operation | Complexity |
-|---|---|
-| Range query | O(L × (log N + k)) via `BTreeMap::range()`, already grouped by label |
-| Exact query | O(m × log N) where m = labels present in target window (via reverse index) |
-| Insert | O(log N) BTreeMap insert per label |
-| CircularBuffer cleanup | O(k × m) iterate first k from `BTreeSet` + targeted removals via `window_to_labels` |
-| ReadBased cleanup | O(N + k × m) scan `read_counts` + targeted removals via `window_to_labels` |
-| Space | O(A × L × N × V) — proportional to stored aggregates, not index overhead |
+|-----------|-------|-------|
+| `InternTable` | O(L) per agg_id | Stores each label string once |
+| `label_map` (per epoch) | O(L · N · V) | Primary index across all epochs |
+| `window_to_ids` | O(N · m) | Reverse index, bounded by epoch |
+| `time_ranges` | O(N) per epoch | BTreeSet of distinct windows |
+| `read_counts` | O(N) total | Counts keyed by TimestampRange |
+| **Total** | **O(A · E · L · N · V)** | E bounded by `max_epochs` (default 4); dominated by label_map |
+
+Arc-sharing means query results reference aggregate objects already in the store — no deep copies on read paths.
+
+---
 
 ## Query Mechanics
 
 ### Range Query
 
-For a query with `[start, end]`:
+For a query `[start, end]`:
 
-1. For each label in `label_map`, use `btree.range((start, 0)..=(end, u64::MAX))` to find candidate entries in O(log n)
-2. Filter by `range_end <= end` (BTreeMap range only bounds `range_start`)
-3. Results are already in chronological order (BTreeMap iteration order) and grouped by label
-4. Update `read_counts` via the `time_ranges` secondary index
+1. Acquire **read lock** on `StoreKeyData` (concurrent queries run in parallel)
+2. For each epoch in `epochs.values()`:
+   - Skip if `min_tr.0 > end || max_tr.1 < start` (epoch range check, O(1) via BTreeSet first/last)
+   - For each label in `label_map`, call `btree.range((start, 0)..=(end, u64::MAX))`, filter `tr.1 <= end`
+   - Stream results directly into a `MetricBucketMap` (grouped by MetricID, no intermediate flat vec)
+3. Resolve MetricIDs → label strings in one pass via `InternTable`
+4. Lock inner `Mutex` briefly to update `read_counts`
 
 ### Exact Query
 
 For exact match `(exact_start, exact_end)`:
 
-1. Use `window_to_labels` to get labels that actually have that window
-2. For those labels only, use `btree.get(&(exact_start, exact_end))` for O(log n) lookup
-2. Results are already grouped by label
+1. Acquire **read lock**
+2. Iterate epochs newest-first (`epochs.values().rev()`):
+   - Use `window_to_ids.get(&range)` to get the sorted `Vec<MetricID>` of labels with that window
+   - For each MetricID, use `label_map[id].get(&range)` — O(log N) point lookup
+   - Stop at the first epoch that has the window (break after first match)
+3. Resolve MetricIDs → labels, update `read_counts`
+
+---
 
 ## Cleanup Policies
 
 ### CircularBuffer
 
-Retains the newest `configured_limit * 4` time ranges:
+Epoch-based eviction — O(1) amortized:
 
-1. Check `time_ranges.len()` against the retention limit
-2. Iterate `time_ranges` from the start (oldest first, already sorted by BTreeSet)
-3. Remove excess entries from `time_ranges`, `read_counts`, and reverse index
-4. Remove from only affected label BTrees using `window_to_labels` membership
+1. On first insert, set `epoch_capacity` from `num_aggregates_to_retain`
+2. After each item insert, call `maybe_rotate_epoch()`:
+   - If current epoch's `window_count() >= epoch_capacity`, open a new epoch (`current_epoch_id + 1`)
+   - If `epochs.len() > max_epochs`, pop the oldest epoch (BTreeMap first entry) — O(1) drop of entire epoch
+   - Purge dropped epoch's windows from `read_counts`
 
 ### ReadBased
 
-Removes entries that have been read `>= threshold` times:
+Read-count triggered eviction:
+
+1. Scan `read_counts` for windows with `count >= threshold`
+2. For each such window, call `EpochData::remove_windows()`:
+   - Remove from `time_ranges`, `window_to_ids`, and only the affected label BTrees (via sorted `Vec<MetricID>` from reverse index)
+3. Drop any epochs that are now empty; re-create `current_epoch_id` entry if it was dropped
+
+### NoCleanup
+
+No eviction — data accumulates indefinitely.
 
-1. Scan `read_counts` for entries meeting the threshold
-2. Remove from `read_counts`, `time_ranges`, and reverse index
-3. Remove from only affected label BTrees using `window_to_labels` membership
+---
 
 ## Concurrency (Per-Key Store)
 
-The per-key store uses a read-lock optimization:
+| Operation | Lock acquired |
+|-----------|--------------|
+| **Insert** | `DashMap` shard lock (briefly) → `RwLock::write` for the duration of the batch |
+| **Range/Exact query** | `DashMap` shard lock (briefly) → `RwLock::read` (concurrent queries run in parallel) → `Mutex::lock` on `read_counts` (briefly, while holding read lock) |
+| **Cleanup** | Runs under the existing write lock; accesses `read_counts` via `Mutex::get_mut()` (no lock overhead — `&mut self` guarantees exclusivity) |
 
-- **Insert**: Acquires a write lock on the `RwLock` (exclusive access needed for `label_map` and `time_ranges`)
-- **Query**: Acquires a read lock on the `RwLock` (multiple queries can run concurrently). Updates `read_counts` by briefly locking the inner `Mutex`
-- **Cleanup**: Runs during insert (under write lock), accesses `read_counts` via `Mutex::get_mut()` (no lock needed since `&mut self` guarantees exclusive access)
+Multiple readers per aggregation_id can proceed concurrently. Writers only block readers of the same aggregation_id, not other aggregation_ids.