add series hash in metrics ingestion, bounded streaming merge#1648
Merged
Conversation
Contributor
|
Important Review skippedThis PR was authored by the user configured for CodeRabbit reviews. CodeRabbit does not review PRs authored by this user. It's recommended to use a dedicated user account to post CodeRabbit review feedback. ⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughAdds deterministic per-series hashing for OTEL metrics, refactors partitioned query streaming to use a bounded Tokio mpsc channel with per-partition forwarders, and conditionally sorts parquet batches for OTel metrics by metric_name then time_partition. ChangesMetrics, Query Streaming, and Parquet Sorting
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (2)
src/otel/metrics.rs (1)
89-91: 💤 Low valueConsider caching the known-fields set to avoid per-call allocation.
OTEL_METRICS_KNOWN_FIELD_LISTis a compile-time constant, but a newHashSetis built on everycompute_series_hashinvocation. Under high ingestion throughput (many data points per second), this repeated allocation adds overhead.♻️ Suggested optimization using `LazyLock`
+use std::sync::LazyLock; + +static KNOWN_FIELDS: LazyLock<std::collections::HashSet<&'static str>> = LazyLock::new(|| { + OTEL_METRICS_KNOWN_FIELD_LIST.iter().copied().collect() +}); + fn compute_series_hash(dp: &Map<String, Value>) -> u64 { - let known: std::collections::HashSet<&str> = - OTEL_METRICS_KNOWN_FIELD_LIST.iter().copied().collect(); let mut label_pairs: Vec<(&str, String)> = dp .iter() - .filter(|(k, _)| !known.contains(k.as_str())) + .filter(|(k, _)| !KNOWN_FIELDS.contains(k.as_str())) .map(|(k, v)| {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/otel/metrics.rs` around lines 89 - 91, The function compute_series_hash repeatedly builds a HashSet from the compile-time OTEL_METRICS_KNOWN_FIELD_LIST on every call; replace that per-call allocation by creating a cached, lazily-initialized HashSet (e.g., via std::sync::OnceLock/OnceCell/LazyLock or once_cell::sync::Lazy) and use the static cached set inside compute_series_hash so the HashSet is constructed once and reused across invocations.src/query/mod.rs (1)
322-333: ⚡ Quick winSpawned task panics would be silently absorbed, potentially losing partition data.
The
JoinHandlereturned bytokio::spawnis dropped immediately, meaning if a task panics (e.g., due to an unexpected error in the stream), that partition's remaining data is silently lost with no indication to the consumer.Consider either:
- Storing
JoinHandles and awaiting them (complex, changes return semantics), or- Using
.abort_handle()and adding panic=abort logging, or- At minimum, wrapping the async block body with
catch_unwindto send an error through the channel before exiting.Option: Catch panics and convert to DataFusion errors
tokio::spawn( async move { + let result = std::panic::AssertUnwindSafe(async { let mut stream: SendableRecordBatchStream = Box::pin(wrapped); use futures::StreamExt; while let Some(batch) = stream.next().await { if tx.send(batch).await.is_err() { break; } } + }); + if let Err(e) = futures::FutureExt::catch_unwind(result).await { + let msg = e.downcast_ref::<&str>().map(|s| s.to_string()) + .or_else(|| e.downcast_ref::<String>().cloned()) + .unwrap_or_else(|| "partition task panicked".to_string()); + let _ = tx.send(Err(datafusion::error::DataFusionError::Execution(msg))).await; } } .instrument(span), );
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/otel/metrics.rs`:
- Around line 89-91: The function compute_series_hash repeatedly builds a
HashSet from the compile-time OTEL_METRICS_KNOWN_FIELD_LIST on every call;
replace that per-call allocation by creating a cached, lazily-initialized
HashSet (e.g., via std::sync::OnceLock/OnceCell/LazyLock or
once_cell::sync::Lazy) and use the static cached set inside compute_series_hash
so the HashSet is constructed once and reused across invocations.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: f5871794-e60a-465b-9cd2-470e7a2f13c7
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (3)
Cargo.tomlsrc/otel/metrics.rssrc/query/mod.rs
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/otel/metrics.rs (1)
89-91: 💤 Low valueConsider caching the known-fields HashSet to avoid per-call allocation.
OTEL_METRICS_KNOWN_FIELD_LISTis a compile-time constant, but theHashSetis rebuilt on everycompute_series_hashcall. For high-throughput ingestion this adds allocation overhead per data point.♻️ Proposed optimization using `std::sync::LazyLock`
+use std::sync::LazyLock; +use std::collections::HashSet; +static KNOWN_FIELDS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| { + OTEL_METRICS_KNOWN_FIELD_LIST.iter().copied().collect() +}); + fn compute_series_hash(dp: &Map<String, Value>) -> u64 { - let known: std::collections::HashSet<&str> = - OTEL_METRICS_KNOWN_FIELD_LIST.iter().copied().collect(); let mut label_pairs: Vec<(&str, String)> = dp .iter() - .filter(|(k, _)| !known.contains(k.as_str())) + .filter(|(k, _)| !KNOWN_FIELDS.contains(k.as_str()))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/otel/metrics.rs` around lines 89 - 91, compute_series_hash currently rebuilds a HashSet from OTEL_METRICS_KNOWN_FIELD_LIST on every call; change this to use a lazily-initialized static cached HashSet (e.g. std::sync::LazyLock or once_cell::sync::Lazy) so the set is constructed once and reused; update compute_series_hash to reference the static (e.g. KNOWN_FIELDS_SET) instead of creating a local HashSet, keeping the rest of the hashing logic unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/otel/metrics.rs`:
- Around line 89-91: compute_series_hash currently rebuilds a HashSet from
OTEL_METRICS_KNOWN_FIELD_LIST on every call; change this to use a
lazily-initialized static cached HashSet (e.g. std::sync::LazyLock or
once_cell::sync::Lazy) so the set is constructed once and reused; update
compute_series_hash to reference the static (e.g. KNOWN_FIELDS_SET) instead of
creating a local HashSet, keeping the rest of the hashing logic unchanged.
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
src/otel/metrics.rs (1)
93-96: ⚡ Quick winConsider caching the known-fields HashSet to avoid per-call allocation.
compute_series_hashis invoked for every data point in the ingestion hot path. Rebuilding the 37-elementHashSeton each call adds unnecessary allocation overhead.♻️ Suggested refactor using LazyLock
+use std::sync::LazyLock; +use std::collections::HashSet; + +static KNOWN_FIELDS: LazyLock<HashSet<&'static str>> = LazyLock::new(|| { + OTEL_METRICS_KNOWN_FIELD_LIST.iter().copied().collect() +}); + fn compute_series_hash(dp: &Map<String, Value>) -> u64 { - let known: std::collections::HashSet<&str> = - OTEL_METRICS_KNOWN_FIELD_LIST.iter().copied().collect(); let mut label_pairs: Vec<(&str, String)> = dp .iter() - .filter(|(k, _)| !known.contains(k.as_str())) + .filter(|(k, _)| !KNOWN_FIELDS.contains(k.as_str())) .map(|(k, v)| {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/otel/metrics.rs` around lines 93 - 96, compute_series_hash currently builds a HashSet from OTEL_METRICS_KNOWN_FIELD_LIST on every call causing allocation overhead; instead create a single static cached HashSet (e.g., via std::sync::LazyLock/OnceLock or once_cell::sync::Lazy) that is initialized once and reused, then update compute_series_hash to reference that static (named e.g., KNOWN_FIELDS_SET) rather than recreating the set each invocation so the per-datapoint hot path avoids repeated allocations.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/otel/metrics.rs`:
- Around line 93-96: compute_series_hash currently builds a HashSet from
OTEL_METRICS_KNOWN_FIELD_LIST on every call causing allocation overhead; instead
create a single static cached HashSet (e.g., via std::sync::LazyLock/OnceLock or
once_cell::sync::Lazy) that is initialized once and reused, then update
compute_series_hash to reference that static (named e.g., KNOWN_FIELDS_SET)
rather than recreating the set each invocation so the per-datapoint hot path
avoids repeated allocations.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/parseable/streams.rs`:
- Around line 676-682: The SortOptions for the time partition currently sets
nulls_first: false while the parquet_writer_props SortingColumn metadata
advertises nulls_first: true, causing a mismatch; update the SortOptions
construction used for time_partition (the SortColumn with values:
batch.column(time_idx)) to set nulls_first: true so it matches the
parquet_writer_props/SortingColumn metadata and keep SortColumn/SortOptions and
parquet_writer_props consistent.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 15ca5235-135d-432e-a3c6-ac7f121f410b
📒 Files selected for processing (1)
src/parseable/streams.rs
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
New Features
Refactor
Chores