MarcusKainth commented on code in PR #4204:
URL: https://github.com/apache/iggy/pull/4204#discussion_r4058294791
##########
core/connectors/sinks/clickhouse_sink/src/body.rs:
##########
@@ -98,7 +98,10 @@ pub(crate) fn build_string_body(
let mut buf = Vec::with_capacity(messages.len() * 64);
for msg in messages {
match &msg.payload {
- Payload::Text(s) => {
+ // `Payload::Proto` holds proto text, so a passthrough body takes
+ // it the same way it takes `Payload::Text`. The JSON and RowBinary
+ // builders above cannot: they need a parsed document.
Review Comment:
I agree on the JSON builder, of all three this was the loudest. Both
builders now parse `Payload::Proto` and take the existing path on success. Text
that is not JSON is skipped, with an accompanying log line, so the existing CSV
and TSV passthrough cases from `raw_to_protobuf` are unaffected. One nuance:
`pretty_json` defaults to false in `proto_convert.rs` `L75`, so the default
fallback text is already one line. I have still gone with parse and
re-serialise as you suggested, since it also validates the document before
ClickHouse sees it.
One correction on RowBinary. It did not lose a failure signal.
`build_row_binary_body` on `master` is byte-identical to this branch, and it
never failed on a variant mismatch. On `master` the reparsed JSON reached
`serialize_row` and was inserted, or failed the batch only on a table mismatch.
The regression was data loss rather than a lost error, and the fix is the same
either way.
One item I want to flag which matches the behaviour on `master`. The string
passthrough writes text verbatim, so a pipeline configured with `pretty_json =
true` and `string_format = "json_each_row"` produces a multi-line JSON document
which ClickHouse will reject. A `text` stream carrying pretty JSON has the same
exposure on `master`; the `Payload::Proto` arm added in this PR makes it
reachable from a `proto_convert` pipeline. I have added documentation in the
`README.md` and left both arms as is, unless you'd rather I re-serialise
`Payload::Text`? I decided against this because it rewrites operator bytes on
master's path. Let me know if you want this resolved within this PR, and if so,
how.
Fixed in `30216eb`. Unit tests cover both builders and the string arm, and a
new `clickhouse/proto_text` integration test on the default JSONEachRow format
loads zero rows and counts zero errors on `d8a8ece`. The `master` vs `HEAD`
matrix is in the summary comment.
##########
core/connectors/sinks/doris_sink/src/lib.rs:
##########
@@ -1098,8 +1098,9 @@ impl Sink for DorisSink {
//
// The lone hard-abort is a non-JSON payload (via `?`): a stream-wide
// schema-contract violation, not a transient chunk failure. Under the
- // documented `schema = "json"` config the SDK drops non-JSON before
- // consume() is called, so this stands as a defensive guard.
+ // documented `schema = "json"` config the runtime's JSON decoder drops
+ // non-JSON bytes before consume() is called, so this stands as a
+ // defensive guard unless a format-converting transform is configured.
Review Comment:
I agree, the comment edit named the hole and then left it open. Doris now
parses proto text per chunk and keeps the abort for text that is not JSON, so
the schema-contract guard still fires for the `text_to_protobuf` and
`raw_to_protobuf` outputs in `proto_convert.rs` `L695` and `L714`. Delta takes
the same parse and keeps its batch error.
One correction on Iceberg. It is not the same shape as Doris and Delta.
`router/mod.rs` `L149-169` is a `filter_map` with a `warn!`, so it silently
dropped the non-JSON rows and only failed when every row was non-JSON, which is
worse, and there is a second site in `router/dynamic_router.rs` `L85-96` where
the route field returns `None`. Both now parse proto text. The silent-drop
shape itself predates this PR and I have left it as is.
Fixed in `30216eb`, with unit tests per sink and a `doris/proto_text`
integration test that loads zero rows on `d8a8ece`. Two things worth having on
record. No decoder produces `Payload::Proto` (`decoders/proto.rs` only emits
`Json` and `Raw`), so every one of these rows is transform-only exposure. And
`master` had its own silent loss on the same configuration once a descriptor is
set: `json_to_protobuf` returns binary `Payload::Raw` at `proto_convert.rs`
`L646`, the `json` tag sent it through `simd_json::to_owned_value`, and the SDK
dropped it before `consume()` ran (`sdk/src/sink.rs` `L203-214`). This branch
tags it `Raw` and delivers it. That row is in the matrix too.
##########
core/connectors/runtime/src/sink.rs:
##########
@@ -674,6 +680,10 @@ async fn process_messages(
continue;
};
+ // Read the tag off the payload before `try_into_vec` consumes it. The
+ // decoder's own schema names the format it reads, not the variant it
+ // returned, and a transform may have changed the variant since.
+ let schema = message.payload.schema();
Review Comment:
Correct on both. `flat_buffer` (`flatbuffer.rs` `L42`) and `proto`
(`proto.rs` `L48`) extract to JSON under the runtime's defaults the same way
`avro` does, so all three change tag, and the section now says so, including
that a `proto` stream into Doris improves. I have added the non-object trigger
as well: `encode_json_with_schema` returns `Err` for any top-level array or
scalar at `proto_convert.rs` `L285`, so a fully configured pipeline takes the
fallback for those messages too. The section is rewritten in the PR body.
##########
core/connectors/sinks/README.md:
##########
@@ -249,6 +249,8 @@ for message in messages {
While the schema of messages (that will be consumed from the Iggy stream),
cannot be controlled by the sink connector itself, the built-in configuration
allows to decide what's the expected format of the messages (the particular
`StreamDecoder` will be used).
+`messages_metadata.schema` names the variant each `Payload` in the batch
actually holds, which is not always the stream's `schema` setting. A decoder
may return a different form than the wire format it reads: with `schema =
"avro"` the Avro decoder extracts to JSON by default, so the batch arrives as
`Payload::Json`. A configured transform can change the variant again. Match on
the `Payload` itself and treat `messages_metadata.schema` as a description of
what arrived, not of how the stream was configured.
Review Comment:
Added. The paragraph now states three things. A poll can arrive as several
`consume()` calls, one per contiguous variant run, and every call repeats the
same `current_offset`. That offset is the partition's high-water offset from
the poll (`sink.rs` `L344`) and was batch-scoped on `master` too, so repeating
it per run is the correct contract rather than a new one, though a sink keying
output on it has to tolerate the repeat. And an empty batch keeps the stream's
configured schema. `e58b1cf`.
##########
core/connectors/sdk/src/sink.rs:
##########
@@ -196,7 +198,12 @@ impl<T: Sink + std::fmt::Debug> SinkContainer<T> {
}
};
- let payload = match
messages_metadata.schema.try_into_payload(message.payload) {
+ // The runtime tags each run from `Payload::schema`, so the
+ // tag names a variant here rather than a wire format.
+ let payload = match Payload::try_from_schema(
Review Comment:
I agree, and this is the finding I most wanted a second pair of eyes on. No
FFI signature changed and no variant was added, but the `Proto` discriminant
changed meaning. The skew is narrower than it first looks and worth stating
precisely: a plugin built against 0.4.0 under this runtime is better off on
`avro`, `flat_buffer` and `proto` streams, because it now receives a `json` tag
it can read, and worse off on exactly one pipeline, a `json` stream with
`proto_convert`, where it rebuilds the `proto`-tagged run as `Payload::Raw`
where `master` handed it `Payload::Json`. That one pipeline is enough to want
the boundary marked. Bumped to 0.5.0 in `1742d52`. The rebuild requirement is
stated in the PR body's Compatibility section, and the connector-sdk skill now
lists a `Schema` variant that changes meaning alongside a new variant as a
breaking change.
On the parse, it is real and now documented, but it comes from the runtime
tag rather than the SDK swap. The `Json` arm of `Payload::try_from_schema` is
character-identical to `Schema::try_into_payload`'s (`lib.rs` `L189-191` and
`L296-298`); the arm that runs changed because the tag did. On `master` the
runtime paid `simd_json::to_vec` and then handed the plugin a `Payload::Avro`
holding JSON bytes it could not use, so the cost was only avoided by being
broken.
##########
core/connectors/sdk/README.md:
##########
@@ -175,7 +177,7 @@ These are SDK configuration fields, not Random or Stdout
`plugin_config` keys. T
#### Transform Options
- **`proto_convert`**: Transform for converting between protobuf and other
formats
-- **`source_format`** / **`target_format`**: Formats to convert between - any
schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`)
+- **`source_format`** / **`target_format`**: Formats to convert between - any
schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`).
`source_format` must match the variant the decoder or a preceding transform
actually produced, or the message is rejected
Review Comment:
Both right. Only `flatbuffer_convert` has the guard (`flatbuffer_convert.rs`
`L114`), and the README's own example at `L128` relies on the absence of one in
`proto_convert`. I have limited the sentence to `flatbuffer_convert` and added
a line that chain order is not defined, because `TransformsConfig` is a
`HashMap` (`configs/connectors.rs` `L250-254`). Changing that to an ordered map
is a separate behaviour change and I think it deserves its own issue rather
than folding it in here. Happy to raise it if you agree. `e58b1cf`.
##########
core/connectors/sinks/elasticsearch_sink/src/lib.rs:
##########
@@ -381,7 +381,9 @@ impl Sink for ElasticsearchSink {
}
}
}
- Payload::Text(text) => simd_json::json!({
+ // `Payload::Proto` holds proto text, so it indexes the
+ // same way `Payload::Text` does.
+ Payload::Text(text) | Payload::Proto(text) =>
simd_json::json!({
Review Comment:
I agree, and the arm was also inconsistent with its `Raw` sibling three
lines up, which already parses first (`elasticsearch_sink/src/lib.rs`
`L372-383`). Both sinks now try the parse and only fall back to the text arm
for proto text that is not JSON. Quickwit had the same flattening before this
PR (`quickwit_sink/src/lib.rs` `L346-349`), so I have fixed it in its own
commit, `c858418`, for consistency. Easy to drop if you would rather keep this
PR narrower. `30216eb` for the two you raised.
##########
core/connectors/sinks/meilisearch_sink/src/lib.rs:
##########
@@ -360,7 +360,9 @@ impl MeilisearchSink {
]),
}
}
- Payload::Text(text) => Map::from_iter([
+ // `Payload::Proto` holds proto text, so it indexes the same
+ // way `Payload::Text` does.
+ Payload::Text(text) | Payload::Proto(text) => Map::from_iter([
Review Comment:
Same as the Elasticsearch arm, fixed in the same commit (`30216eb`): the
parse runs first and the text arm is only the fallback for proto text that is
not JSON.
--
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]
