Next steps: sketch protocol PoC, OTAP metrics integration, live reconfig, multi-topic/multi-node, distributed shuffle

[zzylol]

Context

Next-steps discussion from Slack (Milind, 2026-08-24):

1. PoC of sketch creation, merging, and estimation dataflow processors with the specified control plane configs
2. Implement sketch protocol on top of the self-describing sketch format, and integrate with OTAP metrics
3. Start with single node, multiple topics; test multiple nodes later
4. Add support for live reconfiguration scoped to a single dataflow processor (currently pipeline-wide in OTAP)
5. Distributed shuffle (repartitioning sketches across hosts by a different label than they were originally grouped by, e.g. re-key host×process sketches by process) is not supported in OTAP yet, though most of the underlying functionality (associative sketch merge) already exists.

Status

Two unreviewed PRs already cover most of items (1) and the "self-describing format" half of (2):

• otap: implement Schema/Dictionary/Record codec from data_model.md #5 feat/schema-dictionary-record-codec — implements the SCHEMA/DICTIONARY/RECORD codec (otap::dictionary::SeriesDictionary/SeriesDictionaryDecoder) per docs/data_model.md. Proven in-process only (examples/sketch_pipeline_demo.rs).
• otap: real plugin integration + Arrow IPC network transport #6 feat/otap-plugin-and-network-transport — stacked on otap: implement Schema/Dictionary/Record codec from data_model.md #5. Adds Arrow-IPC-over-TCP transport (otap::wire) and runs the real AsapSketchesPlugin lifecycle in producer and receiver roles over a socket (examples/sketch_producer_node.rs / sketch_receiver_node.rs).

What's still missing, from reading the code directly:

• AsapSketchesProcessor::process() in otap-patch/all/mod.rs:244-266 is a deliberate pass-through — Message::PData is forwarded unchanged, never routed through the plugin. There is no OtapPdata ↔ OtapMetricRecords conversion yet (only OtapMetricRecords ↔ RecordBatch exists, in otap/records.rs).
• Message::Control(NodeControlMsg::Config { .. }) (otap-patch/all/mod.rs:261) is handled with a no-op Ok(()) — never wired to Precompute::update_config, even though that path is already scoped per plugin instance (lifecycle.rs:547), not pipeline-wide.
• Only a no-op StubChannel implements ControlChannel (control_channel.rs) — no real transport (e.g. OpAMP, flagged as the natural fit in docs/data_model.md's "Control plane vs. data plane" section).
• No "topic" concept exists anywhere in the codebase/docs — needs to be defined before item 3 below can be built.
• Nothing in the repo addresses distributed shuffle (item 5) — it's new design work, not just wiring.

Next steps

• Merge otap: implement Schema/Dictionary/Record codec from data_model.md #5 and otap: real plugin integration + Arrow IPC network transport #6. They already deliver most of the sketch creation/merging/estimation PoC; review now to avoid duplicating that work.
• OTAP metrics integration (the "integrate with OTAP metrics" half of item 2 — the remaining, real gap):
  • Write the OtapPdata → OtapMetricRecords conversion (the "one seam left" the README calls out) so otap/records.rs::flatten/lift has something to operate on.
  • Bridge OTAP's per-message process() callback model to AsapSketchesPlugin::start()'s long-running Stream model — e.g. an mpsc channel fed by process(), with plugin.start(...) spawned once at processor construction, and a task draining EmitReceiver back out through effect_handler.
  • Add a role field (producer / receiver) to AsapSketchesUserConfig so create_asap_sketches_processor knows whether to call plugin.start() (raw observations in) or plugin.start_from_envelopes() (upstream SketchStreamBatches in, merge via observe_envelope).
  • Decide which wire encoding OtapPdata carries: the existing Strategy-B encode_batch/decode_batch (otap/encode.rs/decode.rs, disguises sketch data as ordinary OTAP metrics so it can transit generic pipeline hops) vs. otap: implement Schema/Dictionary/Record codec from data_model.md #5's SCHEMA/DICTIONARY/RECORD dictionary economics (currently used by otap: real plugin integration + Arrow IPC network transport #6's otap::wire TCP transport, scoped to asap_sketches→asap_sketches hops).
  • Add a real end-to-end test that runs an actual OTAP pipeline YAML referencing urn:asap:processor:asap_sketches (not just the standalone example binaries) — this is the README's Phase E "cross-host parity" milestone.
• Live reconfiguration, scoped to a single processor (item 4): wire NodeControlMsg::Config (otap-patch/all/mod.rs:261) to Precompute::update_config instead of Ok(()); pick and implement a real ControlChannel transport (OpAMP is the leading candidate per docs/data_model.md) to replace StubChannel.
• Single-node, multiple topics (item 3): define what "topic" maps to in this system (metric name? agg_id? something else?) — confirm with Milind — then drive multiple concurrent series through one producer→receiver pair on one machine, reusing sketch_producer_node/sketch_receiver_node from otap: real plugin integration + Arrow IPC network transport #6 as the harness.
• Multi-node testing: spread producer(s)/receiver(s) across real hosts using the same otap::wire transport once single-node/multi-topic works.
• Distributed shuffle (item 5): design doc for repartition-by-label routing on top of the existing observe_envelope merge primitive — OTAP has no support for this today, so this needs design work before implementation.

[zzylol]

Evaluation against #9 (2026-08-24)

I evaluated each item in this issue against the current head of #9 (5c8402c). PR #9 is open and mergeable, so items attributed to it are implemented on that branch but are not on main until #9 merges.

Executive result

Original item	Result	Evidence / remaining gap
1. Sketch creation, merging, estimation PoC	Substantially complete	Creation/estimation are covered by the runtime and codec suites. Serialized upstream sketch envelopes decode to ObservationValueKind::Envelope and route through Precompute::observe_envelope, with merge coverage in lifecycle/runtime tests. The new live-pipeline test covers scalar ingestion → DDSketch creation → timed emission; it does not yet run a two-processor merge topology.
2. Self-describing sketch protocol + OTAP metrics	Complete in #9	otap::codec directly encodes/decodes SketchEnvelope <-> OtapPdata; the real processor consumes genuine OTAP metrics and emits self-describing sketch envelopes as OTAP metrics. OTAP's native Arrow encoding provides dictionary encoding for metric names and string attributes, so the separate production SeriesDictionary/TCP lane is unnecessary.
3. Single-node multiple topics, then multiple nodes	Not complete	“Topic” still has no agreed mapping (metric_name, agg_id, routing port, or another key). Multiple series already work, but that is not enough to claim multiple topics. No real multi-host test exists.
4. Processor-scoped live reconfiguration	Partially complete	The runtime half is complete in #9: NodeControlMsg::Config applies to the processor instance; compatible changes call Precompute::update_config, algorithm/factory changes drain and rebuild, and window changes recreate the timer. The external control-plane half is not complete: ControlChannel still has only StubChannel; no OpAMP or equivalent transport injects these messages.
5. Distributed shuffle/repartition by label	Not started	Associative envelope merge exists, but there is no routing/repartition design or implementation.

Original checklist, resolved

• Merge otap: implement Schema/Dictionary/Record codec from data_model.md #5 and otap: real plugin integration + Arrow IPC network transport #6. Both are merged into main.
• Choose the production encoding. otap: unify real OtapPdata processor, live reconfiguration, and YAML pipeline E2E #9 chooses direct SketchEnvelope <-> OtapPdata over OTAP's native Arrow path. The legacy SCHEMA/DICTIONARY/RECORD codec remains tested but is not the production otap-engine transport.
• Integrate with real OTAP metrics. AsapSketchesProcessor::process() decodes real OtapPdata, observes scalar metrics or merges embedded envelopes, and emits real OtapPdata through the pipeline effect handler.
• Resolve callback-vs-stream lifecycle. No mpsc bridge is required. OTAP's callback/timer model directly drives the callback-oriented Precompute::observe, tick, and drain APIs.
• Resolve producer/receiver role selection. No role field is required. The decoder distinguishes scalar observations from embedded sketch envelopes; Precompute::observe routes the latter to merge semantics.
• Run a genuine YAML pipeline E2E test. tests/otap_pipeline_e2e.rs parses YAML referencing urn:asap:processor:asap_sketches, builds via OTAP_PIPELINE_FACTORY, runs RuntimePipeline::run_forever, injects real OtapPdata, captures the emitted sketch at an exporter, validates it, and performs runtime shutdown.
• Wire processor-local runtime config handling. Implemented safely in otap::processor.
• Implement a real external control transport. OpAMP remains a candidate; StubChannel is still the only ControlChannel implementation.
• Define and test “multiple topics.” Requires a product/architecture definition before implementation.
• Run a two-processor live merge topology. Existing tests prove merge behavior, but the new real-pipeline test currently contains one asap_sketches processor.
• Run across multiple real hosts / establish cross-host byte parity. The current E2E is a real running pipeline but is in-process and single-host; it should not be labeled cross-host parity.
• Design and implement distributed shuffle. No implementation exists yet.

Verification performed on #9

• cargo test --features otap-engine: 171 passed (114 unit + 11 API + 4 codec + 9 lifecycle + 1 live-pipeline + 32 runtime).
• E2E-specific clippy with -D warnings: clean.
• cargo fmt --check: clean.

Recommended disposition

Merge #9 to land the completed OTAP integration and processor-local reconfiguration work. Keep #7 open as the umbrella until its remaining four deliverables are split or completed:

1. external control transport;
2. topic definition plus multi-topic topology;
3. multi-host/cross-host testing (including a live two-processor merge topology);
4. distributed shuffle design and implementation.

#8 was closed as superseded by #9.