MarcusKainth commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4040014649
##########
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:
Correct, and it is a regression against master, not only a round-trip gap.
Master
tagged with `decoder.schema()`, so a `schema = "json"` stream with a
`proto_convert` transform reached the sink as `Payload::Json`. As pushed for
review (`31c0773`) the same stream reached it as `Payload::Raw`. ClickHouse
skips
those rows and returns success; Delta, Doris and Iceberg fail the batch.
`try_into_payload` had 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. One
function
cannot serve both, so I split them in d8a8ece. `Schema::try_into_payload`
keeps
the source path and its behaviour is unchanged, only its doc comment. The
sink
SDK now rebuilds through `Payload::try_from_schema` (`sdk/src/lib.rs:187`,
called at `sdk/src/sink.rs:203`), the variant-preserving inverse of
`Payload::schema()`, so all six variants come back as the variant they went
in as.
That makes `Payload::Proto` reach a sink for the first time. Elasticsearch
(`elasticsearch_sink/src/lib.rs:386`), Meilisearch
(`meilisearch_sink/src/lib.rs:365`) and the ClickHouse string passthrough
(`clickhouse_sink/src/body.rs:104`) now take it as text, matching Quickwit
and
SurrealDB. ClickHouse JSONEachRow and RowBinary, Delta, Doris and Iceberg
still
will not: proto text is not a JSON document and I would rather they say so
than
sniff. Noted in Compatibility.
One correction on the trigger. `json_to_protobuf` emits `Payload::Proto` only
with no descriptor loaded (`proto_convert.rs:656`) or when the JSON is not a
top-level object (`:285` against `:312`). With a descriptor and an object it
emits `Payload::Raw(binary)`, which does round-trip. There are two further
paths
into `Payload::Proto`: `text_to_protobuf` delegates to the same fallback for
JSON
text (`:693`) and emits it directly for anything else (`:695`), and
`raw_to_protobuf` (`:714`) emits it unconditionally.
Regression tests.
`given_a_payload_when_round_tripped_through_its_own_schema_should_keep_the_variant`
no longer exempts Proto, which `430e5e2` skipped while asserting the
degradation
to `Raw` as intended; it now covers all six variants, with an empty payload
for
each of the five that can carry one. d8a8ece also adds
`given_a_proto_convert_transform_when_the_sink_consumes_should_index_the_payload`,
which drives a `proto_convert` pipeline through a real server, runtime and
Elasticsearch container. It fails without the fix.
##########
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.
--
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]
