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.