otap: implement Schema/Dictionary/Record codec from data_model.md - #5
Merged
Conversation
The OTAP Strategy-B wire codec (encode_batch/decode_batch) emitted one
fully self-describing row per envelope: metric name, labels, sketch_type,
agg_id, schema_version, and encoding all repeated on every RECORD row,
every window, forever -- exactly the anti-pattern docs/data_model.md's
Schema/Dictionary/Record split exists to avoid. Nothing in the codec
implemented the doc's "sent once, referenced by index thereafter"
economics; the design existed only as a field-categorization scheme, not
an implemented protocol behavior.
Add otap::dictionary with SeriesDictionary (sender) / SeriesDictionaryDecoder
(receiver):
- SeriesDictionary::encode assigns stable series_ids and emits SCHEMA /
DICTIONARY / LABELS rows only the first time an agg_id / series is seen;
RECORD always carries just series_id + window bounds + envelope/value.
- SeriesDictionaryDecoder retains SCHEMA/DICTIONARY/LABELS state across a
continuous stream and reconstructs full SketchEnvelopes by joining RECORD
rows back against it, per the doc's own statefulness caveat -- an unknown
series_id/agg_id is a hard decode error, not a silent partial result.
- sketch_size is resolved from PrecomputeConfig::sketch_params via the new
Precompute::active_config() trait method.
- hash_seed/hash_function resolve HashSpec down to the one canonical seed
position (seed_list[canonical_seed_index]) rather than carrying
asap_sketchlib's full 20-entry seed table, per its own self-describing
wire-format doc; SeriesDictionaryDecoder::schema_for() exposes the
resolved value without fabricating a lossy reconstructed HashSpec.
StubPlugin and AsapSketchesPlugin now use this codec for tick/drain
(encode) and inbound-envelope (decode), each with persistent dictionary
state across calls. encode_batch/decode_batch/records::{flatten,lift}
are kept as-is -- they solve a different problem (disguising a payload as
OTAP-Metrics-shaped to transit a generic OTAP pipeline hop) that
docs/data_model.md was never about (its own first line scopes it to the
asap_sketches-to-asap_sketches hop).
Also adds examples/sketch_pipeline_demo.rs: a runnable three-stage demo
(sketch creation processor -> receive processor that merges + queries via
estimate mode -> Prometheus text exposition) showing the dictionary
economics directly in the output -- window 0 sends all four batches,
every later window sends RECORD only.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
11 tasks
zzylol
added a commit
that referenced
this pull request
Aug 24, 2026
Fixes surfaced by code review of #5 (feat/schema-dictionary-record-codec). Files here originate on that branch — cherry-pick onto it if the fix should land in #5 itself rather than only on this stacked branch. - dictionary.rs: SeriesDictionary::identity_key joined agg_id/metric/ labels with unescaped '|'/'='/';' delimiters, so two genuinely different label sets could collide onto the same series_id (e.g. {"a": "1;b=2"} vs. {"a": "1", "b": "2"}). Now length-prefixes each segment, which makes every segment boundary unambiguous regardless of its contents. - dictionary.rs: SeriesDictionaryDecoder::ingest_dictionary unconditionally reset a series' labels to empty on every DICTIONARY row, so a duplicate/replayed row (or a sender that lost its own dictionary state and resent it) would silently wipe out previously- learned labels. Now preserves existing labels via the entry API. - dictionary.rs: resolve_hash_seed always read seed_list[canonical_seed_index], but asap_sketchlib's matrix-family sketches (CountSketch/CountMinSketch) always hash via HashSpec::matrix_seed() (seed_list[0]) on the packed hot path, regardless of canonical_seed_index. Now dispatches on sketch_type. - config.rs: sketch_size_string cast sketch params from f64 to u64 with `as`, silently truncating a misconfigured value (e.g. k = 0.5 truncating to "0") into a fabricated, wrong SCHEMA.sketch_size on the wire. Now validates (finite, non-negative, integral) and omits the field instead of fabricating a value, mirroring resolve_hash_seed's own "omit rather than fabricate" stance. Verified: cargo build/test (otap + default), cargo clippy -D warnings, cargo fmt --check all clean.
zzylol
added a commit
that referenced
this pull request
Aug 24, 2026
…er role)
Implements the "one seam left" the README calls out: putting a
self-describing sketch's serialized bytes onto the wire as a real OTAP
metric, and the reverse (ingesting real OTAP metrics into the
precompute runtime). This is the producer-role half of Phase D/E;
receiver-role (ingesting another asap_sketches node's SketchStreamBatch
as OtapPdata) is a separate, larger piece not covered here.
## What changed
- otap-patch/all/otap_bridge.rs (new): the actual binding.
- Encode (OtapMetricRecords -> OtapPdata): implements the upstream
`MetricsView` trait family (MetricsView/ResourceMetricsView/
ScopeMetricsView/MetricView/DataView/GaugeView/NumberDataPointView/
AttributeView/AnyValueView, plus uninhabited-enum placeholders for
Sum/Histogram/ExponentialHistogram/Summary/Exemplar, which this
binding never produces) as a thin adapter over OtapMetricRecords's
existing flat Arrow schema, then calls upstream's own
`encode_metrics_otap_batch` to build a real `OtapArrowRecords::Metrics`
— pushes all the low-level builder/dictionary-encoding correctness
onto already-tested upstream code instead of hand-rolling it.
- Decode (OtapPdata -> OtapMetricRecords): converts payload to
`OtapArrowRecords` (transparently handles OTLP-proto-bytes or
Arrow-record input via upstream's `TryIntoWithOptions`), walks it
via upstream's own `OtapMetricsView` reader, and rebuilds
OtapMetricRecords's flat 2-batch shape with plain arrow-array
builders (the same pattern records.rs's own `flatten`/`lift` use).
- Scope: only Gauge/Sum (NumberDataPoints) — the scalar shape
Observation/OtapMetricRecords already assume. Histogram/
ExponentialHistogram/Summary rows are skipped and counted
(DecodeOutcome::skipped_non_scalar), not silently dropped.
- otap-patch/all/mod.rs: AsapSketchesProcessor rewritten from a
pass-through into a real processor. Drives a bare `Precompute`
instance directly (obtained via
`AsapSketchesPlugin::from_plugin_config(...).precompute().clone()`,
discarding the plugin wrapper) rather than through
`AsapSketchesPlugin::start()`'s own Tokio-task/Stream lifecycle —
that lifecycle's emit channel now carries `SketchStreamBatch` (PR
#5/#6's dictionary-economics wire format for the asap_sketches ->
asap_sketches transport hop), not the OTAP-Metrics-shaped
`OtapMetricRecords` this adapter needs, so bridging it would need
reconciling formats rather than genuinely fitting. Precompute's own
observe/tick/drain are callback-style already, so OTAP's per-message
`process()` + `effect_handler.start_periodic_timer` (emitting
NodeControlMsg::TimerTick) hosts them directly with no bridging
machinery needed. Also wires NodeControlMsg::Config ->
Precompute::update_config (previously a no-op) — live
reconfiguration for this one processor instance.
- otap-patch/all/Cargo.toml, otap-patch/all/mod.rs,
otap-patch/all/otap_bridge.rs: renamed all OTAP Dataflow crate
imports `otap_df_*` -> `otel_arrow_dfe_*`, matching upstream's very
recent (unreleased as of the pinned commit) package rename,
otel-arrow issue #1848 / .chloggen/otel-arrow-dfe-crate-prefix.yaml.
Revert to `otap_df_*` throughout if the actual internal build pin
predates that rename.
- asap-precompute-rs/{Cargo.toml,Cargo.lock} (prior commit, prerequisite
for this one): Arrow 53 -> 58.3, matching the OTAP workspace's own
pin so RecordBatch is the literal same type across this binding.
## Verification status — read carefully
`otap-patch/` has no standalone build in this repo (confirmed: no
OTAP Dataflow workspace checkout, no lockfile pinning one — see the
repo README's existing "No — depends on the OTAP workspace crates"
note on this directory, which predates this change). Every upstream
type/function signature referenced in otap_bridge.rs and mod.rs was
read directly from a fresh clone of open-telemetry/otel-arrow at
commit 3e85c3460361446ebfce99e9f35fffd2dd5ab740 (2026-08-24) — not
compiled against. The Arrow-only portions (RecordBatch construction,
typed-column accessors) were extracted and compile+test verified in
isolation against the real arrow-array 58.4.0 crate (round-trip test
passing) since that part doesn't depend on the OTAP-specific crates.
The OTAP-specific view-trait implementation (the biggest, riskiest
part) could not be compiled here at all.
Expect a real build pass against the actual pinned OTAP Dataflow
commit to surface mismatches — this is a first cut at the binding,
not a verified-working one. asap-precompute-rs itself (the standalone,
buildable part) is unaffected except for the Arrow version bump, and
remains fully verified: cargo build/test (156/156 passing), clippy
-D warnings, fmt --check all clean.
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
1 participant
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
An audit against
docs/data_model.mdfound that the OTAP Strategy-B wire codec (encode_batch/decode_batch) never actually implemented the doc's Schema/Dictionary/Record economics: every row inlinedmetric,labels,sketch_type,agg_id,schema_version, andencoding, repeated on everyRECORDrow, every window, forever — the exact anti-pattern the doc's rationale exists to avoid. The design only ever existed as a field-categorization scheme, not an implemented protocol behavior.This PR implements it for real.
What changed
otap::dictionarymodule:SeriesDictionary(sender) assigns stableseries_ids and emitsSCHEMA/DICTIONARY/LABELSrows only the first time anagg_id/series is seen;RECORDalways carries justseries_id+ window bounds +envelope/value.SeriesDictionaryDecoder(receiver) retains that state across a continuous stream and reconstructs fullSketchEnvelopes by joiningRECORDrows back against it — an unresolvableseries_id/agg_idis a hard decode error, not a silent partial result, per the doc's own statefulness caveat.sketch_sizeis resolved fromPrecomputeConfig::sketch_paramsvia a newPrecompute::active_config()trait method.hash_seed/hash_functionresolveasap_sketchlib'sHashSpecdown to the one canonical seed position (seed_list[canonical_seed_index]) rather than carrying the full 20-entry seed table — checked againstasap_sketchlib's own self-describing wire-format doc, since oneSCHEMArow (oneagg_id, onesketch_type) only ever needs one seed position.SeriesDictionaryDecoder::schema_for()exposes the resolved value on the receiver side without fabricating a lossy reconstructedHashSpec.StubPluginandAsapSketchesPluginnow use this codec for tick/drain (encode) and inbound-envelope (decode), each with persistent dictionary state across calls.encode_batch/decode_batch/records::{flatten,lift}are unchanged — they solve a different problem (disguising a payload as OTAP-Metrics-shaped so it can transit a generic OTAP pipeline hop) thatdocs/data_model.mdwas never about (its own first line scopes it to theasap_sketches-to-asap_sketcheshop).examples/sketch_pipeline_demo.rs: a runnable three-stage demo — sketch creation processor → receive processor (merges viaobserve_envelope, queries via estimate mode) → Prometheus text exposition. Run withcargo run --example sketch_pipeline_demo --features otap. The printed output makes the dictionary economics directly visible: window 0 sends all four batches, every later window sendsRECORDonly.Testing
cargo test --features otap: 149/149 passing (14 new tests covering first-window-vs-repeat-window batch shapes, distinct series_id assignment, full encode→decode round trips, hash-seed resolution, and decode error handling for unknown series_id/agg_id).cargo fmt --checkandcargo clippy --features otap --tests --examples: clean.cargo run --example sketch_pipeline_demo --features otap: verified end to end — window 0 emitsschema=1 dictionary=1 labels=1 record=1, windows 1-3 emitschema=0 dictionary=0 labels=0 record=1, and the receiver's queried p99 gauge tracks the injected latency drift correctly.🤖 Generated with Claude Code