[adapters] Parse Debezium logical types in the Avro input connector#6620
Conversation
mythical-fred
left a comment
There was a problem hiding this comment.
Nice separation: the generic Coercion layer in coercion.rs stays source-agnostic and delegates every Debezium-specific bit (temporal enum, VariableScaleDecimal parsing) to debezium.rs. Tests cover the wire format for every temporal type (millis/micros/nanos + ISO zoned + org.apache.kafka.connect.data.*), nullable timestamp, wraparound zoned time across midnight, VariableScaleDecimal with a nullable column and a by-name reference, plus a schema-mismatch rejection path. hoist_coercible_types correctly leaves named types alone (they preserve attributes natively) and only rewrites primitives. The time.precision.mode=connect aliases and Date (which stays a plain day count) are handled the way the Debezium docs describe them. Signed-off-by, no AI trailers. Approve.
A couple of small non-blocking observations:
-
Coercion::validatetreats every timestamp coercion as compatible with eitherSqlType::TimestamporSqlType::TimestampTz. That's semantically fine (a plainio.debezium.time.Timestampis UTC epoch millis, so loading it into aTIMESTAMP WITH TIME ZONEcolumn reads it as UTC), but worth a docstring note onis_timestampso future readers don't puzzle over it — especially sinceZonedTimemaps to a non-zonedTIMEcolumn intentionally. -
variable_scale_decimal_to_stringdoesn't reject negativescale. The Debezium spec constrains it to>= 0, andBigDecimal::new(unscaled, scale as i64)will happily interpret a negative scale as a multiplier. Not a real-world concern — Debezium won't emit that — but ascale >= 0check would fail loudly on a corrupted schema-registry payload. -
The
println!("Debezium Avro schema: {schema_str}")indebezium_avro_schema(the helper that only exists in the test file, so no production impact) is debug output that survives PR merges; consider dropping oreprintln!-ing behind a flag.
| Str(String), | ||
| } | ||
|
|
||
| impl Coercion { |
There was a problem hiding this comment.
This file is in a directory called Avro, yet it talks about Kafka.
I can't make a general picture of the code organization.
Moreover, the function name does not mention Kafka at all.
There was a problem hiding this comment.
Kafka Connect (not Kafka) has its own Avro extensions, which is what this PR supports. I agree this is all in bad taste, but we don't control the ecosystem.
| /// Rewrite an Avro schema so that coercion annotations on primitive types | ||
| /// survive parsing by `apache-avro`. | ||
| /// | ||
| /// `apache-avro` discards attributes on primitive schemas, so a `connect.name` |
There was a problem hiding this comment.
Is this a bug in apache-avro? Can we contribute this fix?
There was a problem hiding this comment.
apache-avro is not a well designed crate. I consider this a design bug.
Can we contribute this fix?
Ideally, yes, but this is a time-sensitive feature.
| /// Return an object schema's `connect.name` if it is a recognized coercion on a | ||
| /// primitive type. Named types keep their attributes natively, so they are not | ||
| /// hoisted. | ||
| fn object_hoistable_connect_name(obj: &Map<String, JsonValue>) -> Option<String> { |
There was a problem hiding this comment.
this is making me dizzy
who designed this stuff?
| /// | `io.debezium.time.Timestamp` | `long` milliseconds since epoch | `TIMESTAMP` | | ||
| /// | `io.debezium.time.MicroTimestamp` | `long` microseconds since epoch | `TIMESTAMP` | | ||
| /// | `io.debezium.time.NanoTimestamp` | `long` nanoseconds since epoch | `TIMESTAMP` | | ||
| /// | `io.debezium.time.ZonedTimestamp` | ISO-8601 `string` | `TIMESTAMP` | |
There was a problem hiding this comment.
none of these produces a TIME ZONE?
Note that we don't support TIME WITH TIME ZONE (which I don't think makes sense, but many SQL dialects do), but we do support TIMESTAMP WITH TIME ZONE
There was a problem hiding this comment.
ZonedTimestamp actually deserialized to Timestamp With Timezone. I'll update the docs.
| .ok_or_else(|| "VariableScaleDecimal record is missing the 'value' field".to_string())?; | ||
|
|
||
| let unscaled = BigInt::from_signed_bytes_be(unscaled); | ||
| Ok(BigDecimal::new(unscaled, scale as i64).to_string()) |
There was a problem hiding this comment.
Can't we use our own Decimal? Or is the precision arbitrary?
|
|
||
| /// Debezium message Avro schema with the specified inner record schema. | ||
| fn debezium_avro_schema(value_schema: &str, value_type_name: &str) -> AvroSchema { | ||
| let schema_str = debezium_avro_schema_str(value_schema, value_type_name); |
There was a problem hiding this comment.
Why so many empty lines?
Debezium encodes temporal and variable-scale-decimal columns in forms that
differ from the standard Avro types, tagging them with a `connect.name`
attribute rather than a native Avro logical type.
This commit adds support for several such types:
- Temporal: Time, MicroTime, NanoTime, ZonedTime, Timestamp,
MicroTimestamp, NanoTimestamp, ZonedTimestamp, and the
org.apache.kafka.connect.data.* aliases from time.precision.mode=connect,
parsed into TIME/TIMESTAMP. Date already deserialized correctly as a
plain day count and needs no conversion.
- io.debezium.data.VariableScaleDecimal, a {scale, value} record used for
NUMERIC/DECIMAL columns without a fixed scale, parsed into DECIMAL.
The implementation could have been simpler if it didn't have to fight with
the `apache-avro` crate, which drops attributes on primitive schemas, so these
annotations were lost.
Signed-off-by: Leonid Ryzhyk <ryzhyk@gmail.com>
Debezium encodes temporal and variable-scale-decimal columns in forms that differ from the standard Avro types, tagging them with a
connect.nameattribute rather than a native Avro logical type.This commit adds support for several such types:
The implementation could have been simpler if it didn't have to fight with the
apache-avrocrate, which drops attributes on primitive schemas, so these annotations were lost.Describe Manual Test Plan
Checklist
Breaking Changes?
Mark if you think the answer is yes for any of these components:
Describe Incompatible Changes