Skip to content

fix(query-engine): run top-k CMSHeap on FastPath - #829

Open
milindsrivastava1997 wants to merge 2 commits into
mainfrom
811-engine-run-top-k-cmsheap-on-fastpath-to-match-what-sketch-bench-measures
Open

milindsrivastava1997 wants to merge 2 commits into
mainfrom
811-engine-run-top-k-cmsheap-on-fastpath-to-match-what-sketch-bench-measures

Conversation

@milindsrivastava1997

Copy link
Copy Markdown
Contributor

Closes #811

Runs the query engine top-k CMSHeap directly on CMSHeap<Vector2D, FastPath>.

  • preserves count and rounded integer sum updates
  • removes legacy portable deserialization compatibility
  • accounts for i32 counters in memory estimates
  • adds a fixed-stream FastPath top-k regression test

Tests:

  • cargo test -p query_engine_rust fastpath_fixed_stream_matches_expected_estimates_and_topk --lib
  • cargo test -p query_engine_rust cms_with_heap --lib
  • cargo test -p query_engine_rust --test e2e_precompute_equivalence

@milindsrivastava1997 milindsrivastava1997 left a comment

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Automated code review (Claude Code). Main theme: the f64 → i32 counter switch changes SUM top-k semantics (fractional values dropped, overflow, silent clamping). If the FastPath CMSHeap can only hold integer counts, consider keeping the f64 path for SUM or rejecting non-representable inputs loudly.

fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) {
let weight = if self.count_events { 1.0 } else { value };
self.acc.inner.update(&key.to_semicolon_str(), weight);
let count = if self.count_events {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fractional SUM values lost. Every sample is rounded to i32 and anything ≤0 is dropped by update(). topk(5, sum by (pod)(cpu_seconds)) with gauge samples like 0.3/0.45 → all round to 0 → empty heap, empty result. 1.4 and 1.6 count as 1 and 2, skewing the ranking. The old f64 path got this right.

let count = if self.count_events {
1
} else {
value.round() as i32

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Silent clamping. value.round() as i32 saturates anything above i32::MAX (e.g. a 5e9 bytes sample → 2147483647) and maps NaN (staleness marker) to 0, which is then dropped. Both are wrong results with no signal — violates "Fail loud" (code-design checklist §1).

@@ -758,8 +758,12 @@ impl AccumulatorUpdater for CmsWithHeapAccumulatorUpdater {
}

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Stale doc comment (~line 729, outside the diff): the updater doc still says it wraps CountMinSketchWithHeap and feeds "the sample value itself", but it now rounds to i32 on the FastPath CMSHeap. Checklist §3: docs must change with behavior.

@@ -775,7 +779,7 @@ impl AccumulatorUpdater for CmsWithHeapAccumulatorUpdater {

fn memory_usage_bytes(&self) -> usize {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Memory undercounted. memory_usage_bytes counts only the counter matrix; it ignores the heap (up to heap_size owned String keys plus slot/position indexes). With long semicolon-joined label keys this can rival a 3x1024 i32 matrix, and halving the counter estimate widens the gap for the memory budget.

topk_heap.push(CmsHeapItem { key, value });
pub fn update(&mut self, key: &str, count: i32) {
if count > 0 {
self.inner.insert_many(&DataInput::Str(key), count);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i32 counter overflow. Vector2D adds with plain +=. Once a key (plus its hash collisions) exceeds 2^31 (~2.1 GB of netflow bytes), debug panics and release wraps negative → negative min-estimate, heap evicts the real heavy hitter, query_key returns a negative value. CountMin::merge (increment_by_row) overflows the same way. The old f64 counters had no such limit.

pub fn update(&mut self, key: &str, count: i32) {
if count > 0 {
self.inner.insert_many(&DataInput::Str(key), count);
}

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Non-positive counts silently ignored. update() drops counts ≤0 with no error, so e.g. negative deltas from counter resets disappear and the sketch undercounts while looking healthy. Checklist §1: "Don't let a failure path quietly resolve to a default, an empty value, or a swallowed error."

})
.collect()
}

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Panics on non-string HeapItem, and the match is duplicated in get_topk_keys and serialize_to_json. If another construction/merge path ever puts a non-string key in the heap, a query or serialization crashes the engine thread. Suggest one helper returning Result.

type_name: "CountMinSketchWithHeapAccumulator",
source: e.to_string().into(),
})
Err(SerializationError::Bytes {

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

serialize_to_bytes now always returns Err, and deserialization was removed. Any sink/store calling SerializableToSink::serialize_to_bytes on a boxed top-k accumulator (Kafka/msgpack output, nested serializers) now fails on every top-k window; previously it produced msgpack.

if merged.inner.rows() != acc.inner.rows() || merged.inner.cols() != acc.inner.cols() {
return Err("CountMinSketchWithHeap dimension mismatch".into());
}
merged.inner.merge(&acc.inner);

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Merge result depends on order. Merged heap capacity is now the first accumulator's size instead of min of both: merging (5, 3) gives 5, (3, 5) gives 3, so different window-merge orders can yield different top-k sets. The rewritten test asserts 5 where the old one asserted 3, which hides the change.

Also: the rows/cols pre-check exists only because CountMin::merge panics on mismatch. That validation belongs in sketchlib as a Result-returning merge so other callers are protected too; the new error message also drops the actual dimensions the old one reported.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

engine: run top-k CMSHeap on FastPath to match what sketch-bench measures

1 participant