MarcusKainth commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4040015274
##########
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`
as the reason a batch splits and claims only that a uniform batch stays one
run.
`ProtoConvert` is the one in-tree transform that mixes variants within a
single
configured instance: `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 its JSON is a
top-level object (`:285` against `:312`), so one instance with a descriptor
loaded alternates. Without a descriptor (`:656`) every message falls back
instead, and `text_to_protobuf` (`:693`, `:695`) and `raw_to_protobuf`
(`:714`)
reach `Payload::Proto` by their own routes. The sentence also contradicted
the
second commit on this branch, which caps the per-run reservation because a
batch
can alternate.
`given_mixed_payload_variants_when_batch_is_processed_should_split_into_runs`
already asserted three FFI calls, but drove the mix with a stub transform.
d8a8ece adds two that use the real one.
`given_one_configured_instance_when_objects_and_scalars_are_converted_should_emit_different_payload_variants`
pins that one instance with a loaded descriptor returns `Raw` for an object
and
`Proto` for an array, and
`given_a_proto_convert_transform_when_a_batch_mixes_variants_should_split_into_runs`
runs a real `ProtoConvert` through `process_messages` and asserts three runs
tagged Raw, Proto, Raw with nothing lost.
##########
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, with `Payload::Raw` being the outcome for
any
payload that is not valid `Any` wire bytes.
`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 source plugin sent,
where the
`Any` decode is right. Split in d8a8ece: the sink side now rebuilds through
`Payload::try_from_schema` (`sdk/src/sink.rs:203`), the variant-preserving
inverse of `Payload::schema()`, and the source arm is unchanged. Detail in
the
thread with @rohankumardubey.
One addition to your step 1: the no-descriptor case is not the only one.
With a
descriptor loaded, `encode_json_with_schema` also falls back when the JSON
is not
a top-level object (`proto_convert.rs:285` against `:312`), so the same
instance
can emit `Raw` for one message and `Proto` for the next.
--
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]
