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.
--
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]
