MarcusKainth commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4040015460
##########
core/connectors/sdk/src/lib.rs:
##########
@@ -157,6 +157,26 @@ pub enum Payload {
}
impl Payload {
+ /// The `Schema` describing this payload's variant.
+ ///
+ /// Not the same thing as `StreamDecoder::schema`, which names the wire
+ /// format a decoder reads rather than the variant it hands back: the Avro
+ /// and FlatBuffer decoders return `Payload::Json` whenever
`extract_as_json`
+ /// is set, and the Proto decoder returns `Payload::Json` or `Payload::Raw`
+ /// depending on the path it takes. A transform may change the variant
again
+ /// after that. Anything tagging a payload for transport has to read the
tag
+ /// off the payload it actually holds.
+ pub const fn schema(&self) -> Schema {
+ match self {
+ Payload::Json(_) => Schema::Json,
+ Payload::Raw(_) => Schema::Raw,
+ Payload::Text(_) => Schema::Text,
+ Payload::Proto(_) => Schema::Proto,
Review Comment:
Confirmed, all three steps hold. `Schema::try_into_payload` served two
callers that meant different things by `Schema::Proto`: the sink SDK meant the
payload's variant, `runtime/src/source.rs:609` means the wire format a plugin
sent. Split in d8a8ece, so the sink rebuilds through `Payload::try_from_schema`
and the source arm is untouched. Detail in @rohankumardubey's thread.
One addition to step 1: with a descriptor loaded, `encode_json_with_schema`
also falls back on non-object JSON (`proto_convert.rs:285`), so one instance
can emit both variants.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -599,8 +596,14 @@ async fn process_messages(
}
let decode_elapsed = decode_start.elapsed();
- let mut messages = Vec::with_capacity(decoded.len());
+ // One `Schema` tag covers a whole FFI call, so messages are grouped into
+ // contiguous runs of the same payload variant and each run is sent on its
+ // own. Every decoder and transform in tree is deterministic per instance,
so
+ // a run holds the entire batch in practice and there is one call.
Review Comment:
Agreed, the comment overreached. Dropped in d8a8ece; it now names
`ProtoConvert` and claims only that a uniform batch stays one run.
`json_to_protobuf` returns `Payload::Raw` when `encode_json_with_schema`
succeeds (`proto_convert.rs:646`) and `Payload::Proto` when it does not
(`:652`), discriminated per message by whether the JSON is a top-level object
(`:285`), so one instance with a descriptor alternates. It also contradicted
the branch's second commit, which caps the per-run reservation for exactly that
case.
The existing split test drove the mix with a stub transform. d8a8ece adds
two using the real one: a `proto_convert` unit test pinning both variants from
one instance, and a runtime test running `ProtoConvert` through
`process_messages` asserting three runs tagged Raw, Proto, Raw with nothing
lost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
