From 5b230158b67ca023bf455be45d8b5428ac221e33 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 10 Jul 2026 17:09:52 -0700 Subject: [PATCH] [adapters] Parse Debezium logical types in the Avro input connector 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 --- crates/adapters/src/format/avro.rs | 2 + crates/adapters/src/format/avro/coercion.rs | 388 ++++++++++++ crates/adapters/src/format/avro/debezium.rs | 402 ++++++++++++ .../adapters/src/format/avro/deserializer.rs | 71 ++- crates/adapters/src/format/avro/input.rs | 14 +- crates/adapters/src/format/avro/schema.rs | 19 + crates/adapters/src/format/avro/test.rs | 593 +++++++++++++++++- crates/feldera-types/src/format/avro.rs | 30 + docs.feldera.com/docs/formats/avro.md | 34 + 9 files changed, 1542 insertions(+), 11 deletions(-) create mode 100644 crates/adapters/src/format/avro/coercion.rs create mode 100644 crates/adapters/src/format/avro/debezium.rs diff --git a/crates/adapters/src/format/avro.rs b/crates/adapters/src/format/avro.rs index 178bac15250..1e7cfac0840 100644 --- a/crates/adapters/src/format/avro.rs +++ b/crates/adapters/src/format/avro.rs @@ -5,6 +5,8 @@ use feldera_adapterlib::catalog::AvroSchemaRefs; use feldera_types::format::avro::AvroSchemaRegistryConfig; use schema_registry_converter::blocking::schema_registry::SrSettings; +mod coercion; +mod debezium; pub mod deserializer; pub mod input; pub mod output; diff --git a/crates/adapters/src/format/avro/coercion.rs b/crates/adapters/src/format/avro/coercion.rs new file mode 100644 index 00000000000..38b2117171e --- /dev/null +++ b/crates/adapters/src/format/avro/coercion.rs @@ -0,0 +1,388 @@ +//! Value coercions for the Avro input connector. +//! +//! Some source systems encode a column in a form that differs from the Avro +//! type Feldera expects for the target column. Debezium, for example, encodes a +//! microsecond timestamp as a plain `long` tagged with a `connect.name` +//! attribute, and a variable-scale decimal as a `{scale, value}` record. A +//! [`Coercion`] recognizes such a field and converts its decoded value into a +//! form the target column's deserializer accepts. +//! +//! This module is the generic layer: it identifies coercions from schema +//! metadata, converts values, and validates schemas. The per-type mechanics +//! live in source-specific modules (currently [`super::debezium`]). +//! +//! # Adding a coercion +//! +//! 1. Add a variant to [`Coercion`]. +//! 2. Recognize it in [`Coercion::from_connect_name`] (for Kafka Connect +//! `connect.name` annotations) or extend [`field_coercion`] with a new +//! detector (for other sources). +//! 3. Implement its conversion in [`Coercion::coerce`] and its schema check in +//! [`Coercion::validate`]. +//! 4. If the annotation sits on a primitive Avro type, whose attributes +//! `apache-avro` drops during parsing, [`hoist_coercible_types`] recovers it +//! automatically once step 2 recognizes it. + +use std::collections::BTreeMap; + +use apache_avro::{Schema as AvroSchema, types::Value}; +use feldera_adapterlib::catalog::AvroSchemaRefs; +use feldera_types::program_schema::{ColumnType, SqlType}; +use serde_json::{Map, Value as JsonValue}; + +use super::debezium::{self, DebeziumTimeType}; +use super::schema::{schema_json, schema_unwrap_optional}; +use crate::format::avro::resolve_ref; + +/// `connect.name` of the Debezium variable-scale decimal type. +const DEBEZIUM_VARIABLE_SCALE_DECIMAL: &str = "io.debezium.data.VariableScaleDecimal"; + +/// A value coercion applied to a single Avro field. +/// +/// Identified from schema metadata and applied by the deserializer, a coercion +/// bridges the gap between a field's on-the-wire Avro type and the Feldera +/// column type it feeds. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Coercion { + /// A Debezium temporal type, converted to microseconds. + DebeziumTime(DebeziumTimeType), + /// The `io.debezium.data.VariableScaleDecimal` record, converted to a + /// decimal string. + VariableScaleDecimal, +} + +/// The Serde value a coercion produces. The deserializer dispatches it onto the +/// target column's visitor. +pub enum Coerced { + /// A 64-bit integer, visited as `i64`. + I64(i64), + /// A string, visited as `str`. + Str(String), +} + +impl Coercion { + /// Recognize a coercion from a Kafka Connect `connect.name` annotation. + pub fn from_connect_name(connect_name: &str) -> Option { + if let Some(time_type) = DebeziumTimeType::from_connect_name(connect_name) { + Some(Coercion::DebeziumTime(time_type)) + } else if connect_name == DEBEZIUM_VARIABLE_SCALE_DECIMAL { + Some(Coercion::VariableScaleDecimal) + } else { + None + } + } + + /// A representative name for this coercion, used in error messages. + fn name(self) -> &'static str { + match self { + Coercion::DebeziumTime(time_type) => time_type.canonical_connect_name(), + Coercion::VariableScaleDecimal => DEBEZIUM_VARIABLE_SCALE_DECIMAL, + } + } + + /// Convert a decoded Avro value into the form the target column expects. + pub fn coerce(self, value: &Value) -> Result { + Ok(match self { + Coercion::DebeziumTime(time_type) => Coerced::I64(time_type.to_micros(value)?), + Coercion::VariableScaleDecimal => { + Coerced::Str(debezium::variable_scale_decimal_to_string(value)?) + } + }) + } + + /// Check that this coercion can populate `column_type` from `avro_schema`. + pub fn validate( + self, + avro_schema: &AvroSchema, + refs: &AvroSchemaRefs, + column_type: &ColumnType, + ) -> Result<(), String> { + let (avro_schema, _) = schema_unwrap_optional(avro_schema); + + match self { + Coercion::DebeziumTime(time_type) => { + let (expected_sql, matches_sql) = if time_type.is_timestamp() { + ( + "TIMESTAMP", + matches!(column_type.typ, SqlType::Timestamp | SqlType::TimestampTz), + ) + } else { + ("TIME", matches!(column_type.typ, SqlType::Time)) + }; + if !matches_sql { + return Err(self.wrong_column_error(expected_sql, column_type)); + } + + let matches_primitive = match time_type.avro_primitive() { + "int" => avro_schema == &AvroSchema::Int, + "long" => avro_schema == &AvroSchema::Long, + "string" => avro_schema == &AvroSchema::String, + _ => false, + }; + if !matches_primitive { + return Err(format!( + "invalid Avro schema for Debezium type '{}': expected '{}', but found {}", + self.name(), + time_type.avro_primitive(), + schema_json(avro_schema) + )); + } + } + Coercion::VariableScaleDecimal => { + if column_type.typ != SqlType::Decimal { + return Err(self.wrong_column_error("DECIMAL", column_type)); + } + + let resolved = resolve_ref(avro_schema, refs) + .map_err(|name| format!("error resolving Avro schema reference: {name}"))?; + let valid = matches!( + resolved, + AvroSchema::Record(record) + if record.lookup.contains_key("scale") && record.lookup.contains_key("value") + ); + if !valid { + return Err(format!( + "invalid Avro schema for Debezium type '{}': expected a record with 'scale' and 'value' fields, but found {}", + self.name(), + schema_json(resolved) + )); + } + } + } + + Ok(()) + } + + fn wrong_column_error(self, expected_sql: &str, column_type: &ColumnType) -> String { + format!( + "Debezium type '{}' can only be deserialized into a SQL {expected_sql} column, but the column has type '{}'", + self.name(), + column_type.typ + ) + } +} + +/// Detect the coercion, if any, implied by a record field. +/// +/// Annotations on primitive types are recovered from the field's custom +/// attributes, where [`hoist_coercible_types`] places them. Annotations on +/// named types (records, enums) are read from the (ref-resolved) schema, which +/// preserves them natively. +pub fn field_coercion( + custom_attributes: &BTreeMap, + schema: &AvroSchema, + refs: &AvroSchemaRefs, +) -> Option { + if let Some(coercion) = custom_attributes + .get("connect.name") + .and_then(JsonValue::as_str) + .and_then(Coercion::from_connect_name) + { + return Some(coercion); + } + + let (schema, _) = schema_unwrap_optional(schema); + let schema = resolve_ref(schema, refs).ok()?; + let AvroSchema::Record(record) = schema else { + return None; + }; + record + .attributes + .get("connect.name") + .and_then(JsonValue::as_str) + .and_then(Coercion::from_connect_name) +} + +/// 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` +/// nested inside a field's `type` object is lost. This copies each recognized +/// annotation up to its enclosing record field, where the crate preserves it as +/// a field-level custom attribute. Named types (records, enums) keep their +/// attributes natively and are left untouched. +/// +/// Returns `None` if the input is not valid JSON, letting the caller fall back +/// to the original string so the normal schema-parsing error surfaces. +pub fn hoist_coercible_types(schema_json: &str) -> Option { + let mut value: JsonValue = serde_json::from_str(schema_json).ok()?; + hoist_walk(&mut value); + serde_json::to_string(&value).ok() +} + +/// Recursively visit every record schema, hoisting primitive annotations onto +/// its fields. +fn hoist_walk(value: &mut JsonValue) { + match value { + JsonValue::Object(map) => { + if let Some(JsonValue::Array(fields)) = map.get_mut("fields") { + for field in fields.iter_mut() { + if let JsonValue::Object(field_map) = field { + hoist_field(field_map); + } + } + } + for child in map.values_mut() { + hoist_walk(child); + } + } + JsonValue::Array(items) => { + for item in items.iter_mut() { + hoist_walk(item); + } + } + _ => {} + } +} + +/// Copy a coercion annotation from a field's primitive `type` up to the field. +fn hoist_field(field: &mut Map) { + if field.contains_key("connect.name") { + return; + } + if let Some(connect_name) = field.get("type").and_then(type_hoistable_connect_name) { + field.insert("connect.name".to_string(), JsonValue::String(connect_name)); + } +} + +/// Find a hoistable annotation inside a field's `type`, whether the type is a +/// bare object or a nullable union. +fn type_hoistable_connect_name(type_value: &JsonValue) -> Option { + match type_value { + JsonValue::Object(obj) => object_hoistable_connect_name(obj), + JsonValue::Array(variants) => variants.iter().find_map(|variant| match variant { + JsonValue::Object(obj) => object_hoistable_connect_name(obj), + _ => None, + }), + _ => None, + } +} + +/// 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) -> Option { + let type_name = obj.get("type").and_then(JsonValue::as_str)?; + if !is_primitive_type(type_name) { + return None; + } + let connect_name = obj.get("connect.name")?.as_str()?; + Coercion::from_connect_name(connect_name).map(|_| connect_name.to_string()) +} + +fn is_primitive_type(type_name: &str) -> bool { + matches!( + type_name, + "null" | "boolean" | "int" | "long" | "float" | "double" | "bytes" | "string" + ) +} + +#[cfg(test)] +mod test { + use super::*; + + #[test] + fn detects_and_hoists_primitive_annotation() { + // A temporal annotation on a primitive type is hoisted onto the field + // and then recognized. + let schema = r#"{ + "type": "record", + "name": "Row", + "fields": [ + { "name": "ts", "type": { "type": "long", "connect.name": "io.debezium.time.MicroTimestamp" } }, + { "name": "t", "type": ["null", { "type": "long", "connect.name": "io.debezium.time.NanoTime" }] }, + { "name": "e", "type": { "type": "string", "connect.name": "io.debezium.data.Enum" } }, + { "name": "plain", "type": "long" } + ] + }"#; + + let hoisted = hoist_coercible_types(schema).unwrap(); + let avro = AvroSchema::parse_str(&hoisted).unwrap(); + let AvroSchema::Record(record) = avro else { + panic!("expected record schema"); + }; + let refs = AvroSchemaRefs::new(); + + assert_eq!( + field_coercion( + &record.fields[0].custom_attributes, + &record.fields[0].schema, + &refs + ), + Some(Coercion::DebeziumTime(DebeziumTimeType::TimestampMicros)) + ); + assert_eq!( + field_coercion( + &record.fields[1].custom_attributes, + &record.fields[1].schema, + &refs + ), + Some(Coercion::DebeziumTime(DebeziumTimeType::TimeNanos)) + ); + // Non-temporal and plain fields are not coerced. + assert_eq!( + field_coercion( + &record.fields[2].custom_attributes, + &record.fields[2].schema, + &refs + ), + None + ); + assert_eq!( + field_coercion( + &record.fields[3].custom_attributes, + &record.fields[3].schema, + &refs + ), + None + ); + } + + #[test] + fn detects_named_type_annotation_without_hoisting() { + // A record annotation is recognized natively, and the same annotation is + // recognized through a by-name reference (the second field). + let schema = r#"{ + "type": "record", + "name": "Row", + "fields": [ + { "name": "amount", "type": { + "type": "record", + "name": "VariableScaleDecimal", + "namespace": "io.debezium.data", + "fields": [ + { "name": "scale", "type": "int" }, + { "name": "value", "type": "bytes" } + ], + "connect.name": "io.debezium.data.VariableScaleDecimal" + }}, + { "name": "amount_ref", "type": ["null", "io.debezium.data.VariableScaleDecimal"] } + ] + }"#; + + // The record annotation is not hoisted onto the field. + let hoisted = hoist_coercible_types(schema).unwrap(); + let parsed: JsonValue = serde_json::from_str(&hoisted).unwrap(); + assert!(parsed["fields"][0].get("connect.name").is_none()); + + let avro = AvroSchema::parse_str(&hoisted).unwrap(); + let resolved = apache_avro::schema::ResolvedSchema::try_from(&avro).unwrap(); + let refs: AvroSchemaRefs = resolved + .get_names() + .iter() + .map(|(name, schema)| (name.clone(), (*schema).clone())) + .collect(); + let AvroSchema::Record(record) = &avro else { + panic!("expected record schema"); + }; + + for field in &record.fields { + assert_eq!( + field_coercion(&field.custom_attributes, &field.schema, &refs), + Some(Coercion::VariableScaleDecimal), + "field {} not recognized", + field.name + ); + } + } +} diff --git a/crates/adapters/src/format/avro/debezium.rs b/crates/adapters/src/format/avro/debezium.rs new file mode 100644 index 00000000000..3704842bd8c --- /dev/null +++ b/crates/adapters/src/format/avro/debezium.rs @@ -0,0 +1,402 @@ +//! Debezium-specific value conversions used by the Avro input connector. +//! +//! Debezium does not use Avro's native `timestamp-*`/`time-*`/`decimal` logical +//! types. Instead it tags plain Avro types with a `connect.name` attribute that +//! names a Debezium semantic type, e.g.: +//! +//! ```json +//! { "type": "long", "connect.name": "io.debezium.time.MicroTimestamp" } +//! ``` +//! +//! This module maps those semantic types to the representation Feldera's column +//! types expect. The generic machinery that recognizes annotations, drives the +//! deserializer, and validates schemas lives in [`super::coercion`]; this module +//! provides only the Debezium-specific pieces it delegates to. +//! +//! See the [Debezium temporal types documentation][docs] for the wire format of +//! each type. +//! +//! [docs]: https://debezium.io/documentation/reference/stable/connectors/postgresql.html#postgresql-temporal-types + +use apache_avro::{BigDecimal, types::Value}; +use chrono::{DateTime, Timelike, Utc}; +use num_bigint::BigInt; + +/// Number of microseconds in a full day, used to normalize `ZonedTime`. +const MICROS_PER_DAY: i64 = 86_400 * 1_000_000; + +/// A Debezium temporal semantic type carried by a `connect.name` attribute. +/// +/// Covers the `TIME` and `TIMESTAMP` families. `io.debezium.time.Date` and +/// `org.apache.kafka.connect.data.Date` are intentionally omitted: they are +/// plain day counts that already deserialize correctly into a Feldera `DATE` +/// column without any conversion. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DebeziumTimeType { + /// Milliseconds since midnight (`int`). + /// + /// `io.debezium.time.Time`, `org.apache.kafka.connect.data.Time`. + TimeMillis, + /// Microseconds since midnight (`long`). `io.debezium.time.MicroTime`. + TimeMicros, + /// Nanoseconds since midnight (`long`). `io.debezium.time.NanoTime`. + TimeNanos, + /// ISO-8601 time-with-offset string, e.g. `10:15:30+01:00`. + /// `io.debezium.time.ZonedTime`. + ZonedTime, + /// Milliseconds since the Unix epoch (`long`). + /// + /// `io.debezium.time.Timestamp`, `org.apache.kafka.connect.data.Timestamp`. + TimestampMillis, + /// Microseconds since the Unix epoch (`long`). + /// `io.debezium.time.MicroTimestamp`. + TimestampMicros, + /// Nanoseconds since the Unix epoch (`long`). + /// `io.debezium.time.NanoTimestamp`. + TimestampNanos, + /// ISO-8601 timestamp-with-offset string, + /// e.g. `2011-12-03T10:15:30.030431+01:00`. `io.debezium.time.ZonedTimestamp`. + ZonedTimestamp, +} + +impl DebeziumTimeType { + /// Map a `connect.name` attribute value to a temporal type, or `None` if it + /// is not a recognized temporal type. + pub fn from_connect_name(connect_name: &str) -> Option { + Some(match connect_name { + "io.debezium.time.Time" | "org.apache.kafka.connect.data.Time" => Self::TimeMillis, + "io.debezium.time.MicroTime" => Self::TimeMicros, + "io.debezium.time.NanoTime" => Self::TimeNanos, + "io.debezium.time.ZonedTime" => Self::ZonedTime, + "io.debezium.time.Timestamp" | "org.apache.kafka.connect.data.Timestamp" => { + Self::TimestampMillis + } + "io.debezium.time.MicroTimestamp" => Self::TimestampMicros, + "io.debezium.time.NanoTimestamp" => Self::TimestampNanos, + "io.debezium.time.ZonedTimestamp" => Self::ZonedTimestamp, + _ => return None, + }) + } + + /// A representative `connect.name` for this type, used in error messages. + pub fn canonical_connect_name(self) -> &'static str { + match self { + Self::TimeMillis => "io.debezium.time.Time", + Self::TimeMicros => "io.debezium.time.MicroTime", + Self::TimeNanos => "io.debezium.time.NanoTime", + Self::ZonedTime => "io.debezium.time.ZonedTime", + Self::TimestampMillis => "io.debezium.time.Timestamp", + Self::TimestampMicros => "io.debezium.time.MicroTimestamp", + Self::TimestampNanos => "io.debezium.time.NanoTimestamp", + Self::ZonedTimestamp => "io.debezium.time.ZonedTimestamp", + } + } + + /// `true` if this type populates a `TIMESTAMP` column, `false` for a `TIME` + /// column. + pub fn is_timestamp(self) -> bool { + matches!( + self, + Self::TimestampMillis + | Self::TimestampMicros + | Self::TimestampNanos + | Self::ZonedTimestamp + ) + } + + /// The Avro primitive type name that carries this Debezium type on the wire. + pub fn avro_primitive(self) -> &'static str { + match self { + Self::TimeMillis => "int", + Self::TimeMicros | Self::TimeNanos => "long", + Self::ZonedTime => "string", + Self::TimestampMillis | Self::TimestampMicros | Self::TimestampNanos => "long", + Self::ZonedTimestamp => "string", + } + } + + /// Convert a decoded Avro value into microseconds: since the Unix epoch for + /// timestamp types, since midnight for time types. + pub fn to_micros(self, value: &Value) -> Result { + match self { + Self::TimeMillis | Self::TimestampMillis => millis_to_micros(as_i64(value)?), + Self::TimeMicros | Self::TimestampMicros => as_i64(value), + Self::TimeNanos | Self::TimestampNanos => Ok(as_i64(value)? / 1_000), + Self::ZonedTimestamp => parse_zoned_timestamp_micros(as_str(value)?), + Self::ZonedTime => parse_zoned_time_micros(as_str(value)?), + } + } +} + +/// Convert an `io.debezium.data.VariableScaleDecimal` record into a decimal +/// string. +/// +/// The type is used for `NUMERIC`/`DECIMAL` columns without a fixed scale. On +/// the wire it is a record with a `scale` (`int`) and a `value` (`bytes`, the +/// big-endian two's-complement unscaled integer). The decimal string it yields +/// is parsed by the target `DECIMAL` column deserializer. +pub fn variable_scale_decimal_to_string(value: &Value) -> Result { + let Value::Record(fields) = value else { + return Err(format!( + "expected a VariableScaleDecimal record, but found {value:?}" + )); + }; + + let mut scale = None; + let mut unscaled = None; + for (name, field) in fields { + match name.as_str() { + "scale" => scale = Some(record_int(field)?), + "value" => unscaled = Some(record_bytes(field)?), + _ => {} + } + } + + let scale = scale + .ok_or_else(|| "VariableScaleDecimal record is missing the 'scale' field".to_string())?; + let unscaled = unscaled + .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()) +} + +/// Read an integer out of an Avro `int` or `long` value. +fn as_i64(value: &Value) -> Result { + match value { + Value::Int(i) => Ok(*i as i64), + Value::Long(i) => Ok(*i), + other => Err(format!( + "expected an integer value for a Debezium temporal type, but found {other:?}" + )), + } +} + +/// Read a string out of an Avro `string` value. +fn as_str(value: &Value) -> Result<&str, String> { + match value { + Value::String(s) => Ok(s.as_str()), + other => Err(format!( + "expected a string value for a Debezium zoned temporal type, but found {other:?}" + )), + } +} + +/// Read the `scale` field of a VariableScaleDecimal record. +fn record_int(value: &Value) -> Result { + match value { + Value::Int(i) => Ok(*i), + Value::Union(_, inner) => record_int(inner), + other => Err(format!( + "expected an int for the VariableScaleDecimal 'scale' field, but found {other:?}" + )), + } +} + +/// Read the `value` field of a VariableScaleDecimal record. +fn record_bytes(value: &Value) -> Result<&[u8], String> { + match value { + Value::Bytes(bytes) | Value::Fixed(_, bytes) => Ok(bytes), + Value::Union(_, inner) => record_bytes(inner), + other => Err(format!( + "expected bytes for the VariableScaleDecimal 'value' field, but found {other:?}" + )), + } +} + +fn millis_to_micros(millis: i64) -> Result { + millis.checked_mul(1_000).ok_or_else(|| { + format!("millisecond value {millis} overflows when converted to microseconds") + }) +} + +/// Parse an ISO-8601 `ZonedTimestamp` string into microseconds since the epoch. +fn parse_zoned_timestamp_micros(s: &str) -> Result { + let dt = DateTime::parse_from_rfc3339(s.trim()) + .map_err(|e| format!("invalid Debezium ZonedTimestamp '{s}': {e}"))?; + Ok(dt.timestamp_micros()) +} + +/// Parse an ISO-8601 `ZonedTime` string (e.g. `10:15:30+01:00`) into +/// microseconds since midnight, normalized to UTC. +/// +/// A Feldera `TIME` has no timezone, so the offset is folded into the +/// wall-clock time modulo 24 hours. +fn parse_zoned_time_micros(s: &str) -> Result { + let s = s.trim(); + // Reuse the RFC3339 parser by pinning the time to an arbitrary date. The + // date is discarded after normalizing to UTC; only the time of day matters. + let dt = DateTime::parse_from_rfc3339(&format!("1970-01-01T{s}")) + .map_err(|e| format!("invalid Debezium ZonedTime '{s}': {e}"))?; + let time = dt.with_timezone(&Utc).time(); + let micros = + time.num_seconds_from_midnight() as i64 * 1_000_000 + time.nanosecond() as i64 / 1_000; + Ok(micros.rem_euclid(MICROS_PER_DAY)) +} + +#[cfg(test)] +mod test { + use super::*; + use apache_avro::types::Value; + + #[test] + fn connect_name_mapping() { + assert_eq!( + DebeziumTimeType::from_connect_name("io.debezium.time.MicroTimestamp"), + Some(DebeziumTimeType::TimestampMicros) + ); + assert_eq!( + DebeziumTimeType::from_connect_name("org.apache.kafka.connect.data.Timestamp"), + Some(DebeziumTimeType::TimestampMillis) + ); + // Date is handled by the plain-int path, not as a coercion. + assert_eq!( + DebeziumTimeType::from_connect_name("io.debezium.time.Date"), + None + ); + assert_eq!( + DebeziumTimeType::from_connect_name("io.debezium.data.Json"), + None + ); + } + + #[test] + fn numeric_timestamps_to_micros() { + // 2021-01-01T00:00:00Z = 1_609_459_200 s. + let secs = 1_609_459_200i64; + assert_eq!( + DebeziumTimeType::TimestampMillis + .to_micros(&Value::Long(secs * 1_000)) + .unwrap(), + secs * 1_000_000 + ); + assert_eq!( + DebeziumTimeType::TimestampMicros + .to_micros(&Value::Long(secs * 1_000_000)) + .unwrap(), + secs * 1_000_000 + ); + assert_eq!( + DebeziumTimeType::TimestampNanos + .to_micros(&Value::Long(secs * 1_000_000_000)) + .unwrap(), + secs * 1_000_000 + ); + } + + #[test] + fn numeric_times_to_micros() { + // 01:02:03.004 = 3_723_004 ms since midnight. + let millis = 3_723_004i64; + let micros = millis * 1_000; + assert_eq!( + DebeziumTimeType::TimeMillis + .to_micros(&Value::Int(millis as i32)) + .unwrap(), + micros + ); + assert_eq!( + DebeziumTimeType::TimeMicros + .to_micros(&Value::Long(micros)) + .unwrap(), + micros + ); + assert_eq!( + DebeziumTimeType::TimeNanos + .to_micros(&Value::Long(micros * 1_000)) + .unwrap(), + micros + ); + } + + #[test] + fn zoned_timestamp_to_micros() { + // Example from the Debezium docs, with a +01:00 offset. + let micros = DebeziumTimeType::ZonedTimestamp + .to_micros(&Value::String( + "2011-12-03T10:15:30.030431+01:00".to_string(), + )) + .unwrap(); + // Equivalent UTC instant: 2011-12-03T09:15:30.030431Z. + let expected = DateTime::parse_from_rfc3339("2011-12-03T09:15:30.030431Z") + .unwrap() + .timestamp_micros(); + assert_eq!(micros, expected); + } + + #[test] + fn zoned_time_to_micros() { + // 10:15:30+01:00 normalizes to 09:15:30 UTC. + let micros = DebeziumTimeType::ZonedTime + .to_micros(&Value::String("10:15:30+01:00".to_string())) + .unwrap(); + let expected = (9 * 3_600 + 15 * 60 + 30) * 1_000_000; + assert_eq!(micros, expected); + } + + #[test] + fn zoned_time_wraps_across_midnight() { + // 00:30:00+01:00 is 23:30:00 UTC on the previous day; only the time + // of day survives. + let micros = DebeziumTimeType::ZonedTime + .to_micros(&Value::String("00:30:00+01:00".to_string())) + .unwrap(); + let expected = (23i64 * 3_600 + 30 * 60) * 1_000_000; + assert_eq!(micros, expected); + } + + #[test] + fn zoned_time_utc_with_fraction() { + let micros = DebeziumTimeType::ZonedTime + .to_micros(&Value::String("13:37:03.123456Z".to_string())) + .unwrap(); + let expected = (13i64 * 3_600 + 37 * 60 + 3) * 1_000_000 + 123_456; + assert_eq!(micros, expected); + } + + #[test] + fn invalid_zoned_values_error() { + assert!( + DebeziumTimeType::ZonedTimestamp + .to_micros(&Value::String("not-a-timestamp".to_string())) + .is_err() + ); + assert!( + DebeziumTimeType::ZonedTime + .to_micros(&Value::String("99:99:99Z".to_string())) + .is_err() + ); + } + + #[test] + fn variable_scale_decimal() { + let record = |scale: i32, unscaled: i64| { + Value::Record(vec![ + ("scale".to_string(), Value::Int(scale)), + ( + "value".to_string(), + Value::Bytes(BigInt::from(unscaled).to_signed_bytes_be()), + ), + ]) + }; + + assert_eq!( + variable_scale_decimal_to_string(&record(2, 12345)).unwrap(), + "123.45" + ); + assert_eq!( + variable_scale_decimal_to_string(&record(3, -6789)).unwrap(), + "-6.789" + ); + assert_eq!( + variable_scale_decimal_to_string(&record(0, 42)).unwrap(), + "42" + ); + } + + #[test] + fn variable_scale_decimal_missing_field_errors() { + let record = Value::Record(vec![("scale".to_string(), Value::Int(2))]); + assert!(variable_scale_decimal_to_string(&record).is_err()); + } +} diff --git a/crates/adapters/src/format/avro/deserializer.rs b/crates/adapters/src/format/avro/deserializer.rs index 6fa7a5fa43a..df8dc257d4e 100644 --- a/crates/adapters/src/format/avro/deserializer.rs +++ b/crates/adapters/src/format/avro/deserializer.rs @@ -47,6 +47,7 @@ use std::{ use crate::format::avro::resolve_ref; +use super::coercion::{Coerced, Coercion, field_coercion}; use super::input::avro_de_config; fn deserialize_decimal(d: &Decimal, schema: &Schema) -> Result { @@ -70,6 +71,10 @@ pub struct Deserializer<'de> { schema: &'de Schema, refs: &'de AvroSchemaRefs, input: &'de Value, + /// Value coercion for the enclosing record field, if any. When set, the + /// decoded Avro value is converted (see [`Coercion`]) before being handed to + /// the target column's deserializer. + coercion: Option, } struct SeqDeserializer<'de> { @@ -99,6 +104,21 @@ impl<'de> Deserializer<'de> { input, schema, refs, + coercion: None, + } + } + + fn with_coercion( + input: &'de Value, + schema: &'de Schema, + refs: &'de AvroSchemaRefs, + coercion: Option, + ) -> Self { + Deserializer { + input, + schema, + refs, + coercion, } } } @@ -180,6 +200,26 @@ fn unwrap_union<'de>(v: &'de Value, schema: &'de Schema) -> (&'de Value, &'de Sc } } +/// Apply a [`Coercion`] to `value` and hand the result to `visitor`. +/// +/// This is the single point at which coercions produce Serde values, called +/// from every deserialize entry point a coerced field can reach. +// The error type is determined by the `apache_avro` crate. +#[allow(clippy::result_large_err)] +fn visit_coerced<'de, V: Visitor<'de>>( + coercion: Coercion, + value: &Value, + visitor: V, +) -> Result { + match coercion + .coerce(value) + .map_err(|e| de::Error::custom(format!("error applying Avro value coercion: {e}")))? + { + Coerced::I64(i) => visitor.visit_i64(i), + Coerced::Str(s) => visitor.visit_str(&s), + } +} + impl<'de> de::Deserializer<'de> for &'_ Deserializer<'de> { type Error = Error; @@ -188,6 +228,9 @@ impl<'de> de::Deserializer<'de> for &'_ Deserializer<'de> { V: Visitor<'de>, { let (val, schema) = unwrap_union(self.input, self.schema); + if let Some(coercion) = self.coercion { + return visit_coerced(coercion, val, visitor); + } match val { Value::Null => visitor.visit_unit(), &Value::Boolean(b) => visitor.visit_bool(b), @@ -233,7 +276,11 @@ impl<'de> de::Deserializer<'de> for &'_ Deserializer<'de> { where V: Visitor<'de>, { - match unwrap_union(self.input, self.schema).0 { + let (val, _) = unwrap_union(self.input, self.schema); + if let Some(coercion) = self.coercion { + return visit_coerced(coercion, val, visitor); + } + match val { Value::String(s) | Value::Enum(_, s) => visitor.visit_borrowed_str(s), Value::Bytes(bytes) | Value::Fixed(_, bytes) => { let s = @@ -251,7 +298,11 @@ impl<'de> de::Deserializer<'de> for &'_ Deserializer<'de> { where V: Visitor<'de>, { - match unwrap_union(self.input, self.schema).0 { + let (val, _) = unwrap_union(self.input, self.schema); + if let Some(coercion) = self.coercion { + return visit_coerced(coercion, val, visitor); + } + match val { Value::String(s) | Value::Enum(_, s) => visitor.visit_borrowed_str(s), Value::Bytes(bytes) | Value::Fixed(_, bytes) => { let s = String::from_utf8(bytes.to_owned()) @@ -310,13 +361,19 @@ impl<'de> de::Deserializer<'de> for &'_ Deserializer<'de> { self.schema ))); }; - visitor.visit_some(&Deserializer::new( + visitor.visit_some(&Deserializer::with_coercion( inner, &union_schema.variants()[*i as usize], self.refs, + self.coercion, )) } - v => visitor.visit_some(&Deserializer::new(v, self.schema, self.refs)), + v => visitor.visit_some(&Deserializer::with_coercion( + v, + self.schema, + self.refs, + self.coercion, + )), } } @@ -517,10 +574,12 @@ impl<'de> de::MapAccess<'de> for RecordDeserializer<'de> { { match self.value.take() { Some(value) => { - let result = seed.deserialize(&Deserializer::new( + let field = &self.record_schema.fields[self.next_field_index]; + let result = seed.deserialize(&Deserializer::with_coercion( value, - &self.record_schema.fields[self.next_field_index].schema, + &field.schema, self.refs, + field_coercion(&field.custom_attributes, &field.schema, self.refs), )); self.next_field_index += 1; result diff --git a/crates/adapters/src/format/avro/input.rs b/crates/adapters/src/format/avro/input.rs index 381d93fea1c..b320a28d64b 100644 --- a/crates/adapters/src/format/avro/input.rs +++ b/crates/adapters/src/format/avro/input.rs @@ -284,7 +284,19 @@ impl AvroParser { &self, schema_str: &str, ) -> Result<(AvroSchema, AvroSchemaRefs, AvroSchema), ControllerError> { - let schema = AvroSchema::parse_str(schema_str).map_err(|e| { + // Debezium encodes some columns with `connect.name` annotations that + // `apache-avro` drops for primitive types. Hoist them onto their record + // fields so they survive parsing (see `super::coercion`). + let schema_str: Cow = + if matches!(self.config.update_format, AvroUpdateFormat::Debezium) { + super::coercion::hoist_coercible_types(schema_str) + .map(Cow::Owned) + .unwrap_or(Cow::Borrowed(schema_str)) + } else { + Cow::Borrowed(schema_str) + }; + + let schema = AvroSchema::parse_str(&schema_str).map_err(|e| { ControllerError::invalid_parser_configuration( &self.endpoint_name, &format!("error parsing Avro schema: {e}"), diff --git a/crates/adapters/src/format/avro/schema.rs b/crates/adapters/src/format/avro/schema.rs index f4d5c44cc84..d8092395346 100644 --- a/crates/adapters/src/format/avro/schema.rs +++ b/crates/adapters/src/format/avro/schema.rs @@ -13,6 +13,7 @@ use feldera_adapterlib::catalog::AvroSchemaRefs; use feldera_types::program_schema::{ColumnType, Field, Relation, SqlIdentifier, SqlType}; use tracing::warn; +use crate::format::avro::coercion::field_coercion; use crate::format::avro::resolve_ref; /// Indicates whether the field has an optional type (`["null", T]`) and, @@ -82,6 +83,24 @@ pub fn validate_struct_schema( } }; + // A field carrying a coercion annotation (e.g. a Debezium temporal type + // or variable-scale decimal) is validated against the SQL column type by + // the coercion itself, since its Avro schema differs from the column's + // natural Avro type. + if let Some(coercion) = + field_coercion(&avro_field.custom_attributes, &avro_field.schema, refs) + { + coercion + .validate(&avro_field.schema, refs, &field.columntype) + .map_err(|e| { + format!( + "error validating schema for column '{}': {e}", + field.name.name() + ) + })?; + continue; + } + validate_field_schema( &avro_field.name, &avro_field.schema, diff --git a/crates/adapters/src/format/avro/test.rs b/crates/adapters/src/format/avro/test.rs index 98e945d8798..e3f511a3261 100644 --- a/crates/adapters/src/format/avro/test.rs +++ b/crates/adapters/src/format/avro/test.rs @@ -16,11 +16,12 @@ use crate::{catalog::SerBatchReader, util::run_in_posix_runtime}; use apache_avro::{ Schema as AvroSchema, from_avro_datum, schema::ResolvedSchema, to_avro_datum, types::Value, }; +use chrono::{DateTime, Utc}; use dbsp::trace::BatchReaderFactories; use dbsp::typed_batch::{DynSpineSnapshot, SpineSnapshot as TypedSpineSnapshot, TypedBatch}; use dbsp::{DBData, IndexedZSetReader, OrdIndexedZSet, OrdZSet, ZWeight, utils::Tup2}; use feldera_adapterlib::transport::OutputBatchType; -use feldera_sqllib::{ByteArray, Uuid, Variant}; +use feldera_sqllib::{ByteArray, Date, SqlDecimal, Time, Timestamp, TimestampTz, Uuid, Variant}; use feldera_types::{ deserialize_table_record, format::avro::{AvroEncoderConfig, AvroEncoderKeyMode}, @@ -33,6 +34,7 @@ use feldera_types::{ serialize_struct, }; use itertools::Itertools; +use num_bigint::BigInt; use proptest::prelude::*; use proptest::proptest; use rand::rngs::StdRng; @@ -122,6 +124,18 @@ impl DebeziumMessage { /// 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); + + println!("Debezium Avro schema: {schema_str}"); + + AvroSchema::parse_str(&schema_str).unwrap() +} + +/// Raw JSON of the Debezium envelope schema embedding the given inner record +/// schema. Unlike [`debezium_avro_schema`], the string is returned verbatim so +/// that Debezium `connect.name` annotations survive (parsing them into an +/// `AvroSchema` drops attributes on primitive types). +fn debezium_avro_schema_str(value_schema: &str, value_type_name: &str) -> String { // Note: placing `after` before `before` to trigger schema reference resolution. let schema_str = r#"{ "type": "record", @@ -202,9 +216,7 @@ fn debezium_avro_schema(value_schema: &str, value_type_name: &str) -> AvroSchema "connect.name": "test_namespace.Envelope" }"#.replace("VALUE_SCHEMA", value_schema).replace("VALUE_TYPE", value_type_name); - println!("Debezium Avro schema: {schema_str}"); - - AvroSchema::parse_str(&schema_str).unwrap() + schema_str } fn serialize_record(x: &T, schema: &AvroSchema) -> Vec @@ -1063,6 +1075,579 @@ fn test_ms_time() { run_parser_test(vec![test]); } +/// Raw wire representation of a row carrying every Debezium temporal type. +/// +/// The fields hold the exact values Debezium puts on the wire (milliseconds, +/// microseconds, or nanoseconds since epoch/midnight, or ISO-8601 strings) so +/// that the test can encode a realistic Debezium message. The connector must +/// convert these into the [`TestTemporal`] representation on the way in. +#[derive(Debug, Clone, PartialEq, Eq)] +struct TestTemporalRaw { + id: i64, + ts_millis: i64, + ts_micros: i64, + ts_nanos: i64, + ts_zoned: String, + ts_connect: i64, + ts_opt: Option, + t_millis: i32, + t_micros: i64, + t_nanos: i64, + t_zoned: String, + t_connect: i32, + dt: i32, +} + +serialize_table_record!(TestTemporalRaw[13]{ + r#id["id"]: i64, + r#ts_millis["ts_millis"]: i64, + r#ts_micros["ts_micros"]: i64, + r#ts_nanos["ts_nanos"]: i64, + r#ts_zoned["ts_zoned"]: String, + r#ts_connect["ts_connect"]: i64, + r#ts_opt["ts_opt"]: Option, + r#t_millis["t_millis"]: i32, + r#t_micros["t_micros"]: i64, + r#t_nanos["t_nanos"]: i64, + r#t_zoned["t_zoned"]: String, + r#t_connect["t_connect"]: i32, + r#dt["dt"]: i32 +}); + +deserialize_table_record!(TestTemporalRaw["Temporal", Variant, 13] { + (r#id, "id", false, i64, |_| None), + (r#ts_millis, "ts_millis", false, i64, |_| None), + (r#ts_micros, "ts_micros", false, i64, |_| None), + (r#ts_nanos, "ts_nanos", false, i64, |_| None), + (r#ts_zoned, "ts_zoned", false, String, |_| None), + (r#ts_connect, "ts_connect", false, i64, |_| None), + (r#ts_opt, "ts_opt", true, Option, |_| Some(None)), + (r#t_millis, "t_millis", false, i32, |_| None), + (r#t_micros, "t_micros", false, i64, |_| None), + (r#t_nanos, "t_nanos", false, i64, |_| None), + (r#t_zoned, "t_zoned", false, String, |_| None), + (r#t_connect, "t_connect", false, i32, |_| None), + (r#dt, "dt", false, i32, |_| None) +}); + +/// Decoded representation of [`TestTemporalRaw`]: every temporal column parsed +/// into the matching Feldera type. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TestTemporal { + id: i64, + ts_millis: Timestamp, + ts_micros: Timestamp, + ts_nanos: Timestamp, + ts_zoned: TimestampTz, + ts_connect: Timestamp, + ts_opt: Option, + t_millis: Time, + t_micros: Time, + t_nanos: Time, + t_zoned: Time, + t_connect: Time, + dt: Date, +} + +serialize_table_record!(TestTemporal[13]{ + r#id["id"]: i64, + r#ts_millis["ts_millis"]: Timestamp, + r#ts_micros["ts_micros"]: Timestamp, + r#ts_nanos["ts_nanos"]: Timestamp, + r#ts_zoned["ts_zoned"]: TimestampTz, + r#ts_connect["ts_connect"]: Timestamp, + r#ts_opt["ts_opt"]: Option, + r#t_millis["t_millis"]: Time, + r#t_micros["t_micros"]: Time, + r#t_nanos["t_nanos"]: Time, + r#t_zoned["t_zoned"]: Time, + r#t_connect["t_connect"]: Time, + r#dt["dt"]: Date +}); + +deserialize_table_record!(TestTemporal["Temporal", Variant, 13] { + (r#id, "id", false, i64, |_| None), + (r#ts_millis, "ts_millis", false, Timestamp, |_| None), + (r#ts_micros, "ts_micros", false, Timestamp, |_| None), + (r#ts_nanos, "ts_nanos", false, Timestamp, |_| None), + (r#ts_zoned, "ts_zoned", false, TimestampTz, |_| None), + (r#ts_connect, "ts_connect", false, Timestamp, |_| None), + (r#ts_opt, "ts_opt", true, Option, |_| Some(None)), + (r#t_millis, "t_millis", false, Time, |_| None), + (r#t_micros, "t_micros", false, Time, |_| None), + (r#t_nanos, "t_nanos", false, Time, |_| None), + (r#t_zoned, "t_zoned", false, Time, |_| None), + (r#t_connect, "t_connect", false, Time, |_| None), + (r#dt, "dt", false, Date, |_| None) +}); + +impl TestTemporal { + /// Avro value schema covering every supported Debezium temporal type. Each + /// column is a plain `int`/`long`/`string` tagged with a Debezium + /// `connect.name`, exactly as Debezium emits them. + fn value_avro_schema() -> &'static str { + r#"{ + "type": "record", + "name": "Temporal", + "connect.name": "test_namespace.Temporal", + "fields": [ + { "name": "id", "type": "long" }, + { "name": "ts_millis", "type": { "type": "long", "connect.name": "io.debezium.time.Timestamp" } }, + { "name": "ts_micros", "type": { "type": "long", "connect.name": "io.debezium.time.MicroTimestamp" } }, + { "name": "ts_nanos", "type": { "type": "long", "connect.name": "io.debezium.time.NanoTimestamp" } }, + { "name": "ts_zoned", "type": { "type": "string", "connect.name": "io.debezium.time.ZonedTimestamp" } }, + { "name": "ts_connect", "type": { "type": "long", "connect.name": "org.apache.kafka.connect.data.Timestamp" } }, + { "name": "ts_opt", "type": ["null", { "type": "long", "connect.name": "io.debezium.time.MicroTimestamp" }], "default": null }, + { "name": "t_millis", "type": { "type": "int", "connect.name": "io.debezium.time.Time" } }, + { "name": "t_micros", "type": { "type": "long", "connect.name": "io.debezium.time.MicroTime" } }, + { "name": "t_nanos", "type": { "type": "long", "connect.name": "io.debezium.time.NanoTime" } }, + { "name": "t_zoned", "type": { "type": "string", "connect.name": "io.debezium.time.ZonedTime" } }, + { "name": "t_connect", "type": { "type": "int", "connect.name": "org.apache.kafka.connect.data.Time" } }, + { "name": "dt", "type": { "type": "int", "connect.name": "io.debezium.time.Date" } } + ] + }"# + } + + fn schema() -> Vec { + vec![ + Field::new("id".into(), ColumnType::bigint(false)), + Field::new("ts_millis".into(), ColumnType::timestamp(false)), + Field::new("ts_micros".into(), ColumnType::timestamp(false)), + Field::new("ts_nanos".into(), ColumnType::timestamp(false)), + Field::new("ts_zoned".into(), ColumnType::timestamp_tz(false)), + Field::new("ts_connect".into(), ColumnType::timestamp(false)), + Field::new("ts_opt".into(), ColumnType::timestamp(true)), + Field::new("t_millis".into(), ColumnType::time(false)), + Field::new("t_micros".into(), ColumnType::time(false)), + Field::new("t_nanos".into(), ColumnType::time(false)), + Field::new("t_zoned".into(), ColumnType::time(false)), + Field::new("t_connect".into(), ColumnType::time(false)), + Field::new("dt".into(), ColumnType::date(false)), + ] + } + + fn relation_schema() -> Relation { + Relation { + name: SqlIdentifier::new("Temporal", false), + fields: Self::schema(), + materialized: false, + properties: BTreeMap::new(), + primary_key: None, + } + } +} + +/// Parse every Debezium temporal semantic type into the matching Feldera +/// `TIME`/`TIMESTAMP`/`DATE` column. +#[test] +fn test_debezium_temporal_types() { + // Base instant with millisecond precision so its millisecond, microsecond, + // and nanosecond encodings all denote the same point in time. + let instant = DateTime::parse_from_rfc3339("2021-06-15T12:30:45.123Z") + .unwrap() + .with_timezone(&Utc); + let ts_micros = instant.timestamp_micros(); + let ts_millis = ts_micros / 1_000; + let ts_nanos = ts_micros * 1_000; + let dt_days = (instant.timestamp() / 86_400) as i32; + + // Time of day 12:30:45.123, again at millisecond precision. + let time_micros = (12 * 3_600 + 30 * 60 + 45) * 1_000_000 + 123_000; + let time_millis = (time_micros / 1_000) as i32; + let time_nanos = time_micros * 1_000; + let time_as_time = Time::from_nanoseconds((time_micros * 1_000) as u64); + + let raw = TestTemporalRaw { + id: 1, + ts_millis, + ts_micros, + ts_nanos, + // +02:00 wall clock for the same instant, to exercise offset handling. + ts_zoned: "2021-06-15T14:30:45.123+02:00".to_string(), + ts_connect: ts_millis, + ts_opt: Some(ts_micros), + t_millis: time_millis, + t_micros: time_micros, + t_nanos: time_nanos, + t_zoned: "14:30:45.123+02:00".to_string(), + t_connect: time_millis, + dt: dt_days, + }; + + let expected = TestTemporal { + id: 1, + ts_millis: Timestamp::from_microseconds(ts_micros), + ts_micros: Timestamp::from_microseconds(ts_micros), + ts_nanos: Timestamp::from_microseconds(ts_micros), + ts_zoned: TimestampTz::from_microseconds(ts_micros), + ts_connect: Timestamp::from_microseconds(ts_micros), + ts_opt: Some(Timestamp::from_microseconds(ts_micros)), + t_millis: time_as_time, + t_micros: time_as_time, + t_nanos: time_as_time, + t_zoned: time_as_time, + t_connect: time_as_time, + dt: Date::from_days(dt_days), + }; + + // Second row exercises a NULL value in the nullable timestamp column. + let raw_null = TestTemporalRaw { + id: 2, + ts_opt: None, + ..raw.clone() + }; + let expected_null = TestTemporal { + id: 2, + ts_opt: None, + ..expected.clone() + }; + + // Encode a Debezium message from the raw values. The envelope schema is + // parsed here purely to drive Avro encoding; the connector receives the raw + // schema string (below) with the `connect.name` annotations intact. + let envelope_str = debezium_avro_schema_str(TestTemporal::value_avro_schema(), "Temporal"); + let envelope_schema = AvroSchema::parse_str(&envelope_str).unwrap(); + let resolved = ResolvedSchema::try_from(&envelope_schema).unwrap(); + + let encode = |row: &TestTemporalRaw| { + let mut buffer = vec![0; 5]; + let serializer = AvroSchemaSerializer::new(&envelope_schema, resolved.get_names(), true); + let dbz_message = DebeziumMessage::new("u", Some(row.clone()), Some(row.clone())); + let val = dbz_message + .serialize_with_context(serializer, &avro_ser_config()) + .unwrap(); + let mut avro_record = to_avro_datum(&envelope_schema, val).unwrap(); + buffer.append(&mut avro_record); + (buffer, vec![]) + }; + + let test = TestCase { + relation_schema: TestTemporal::relation_schema(), + config: AvroParserConfig { + update_format: AvroUpdateFormat::Debezium, + schema: Some(envelope_str), + skip_schema_id: false, + registry_config: Default::default(), + }, + input_batches: vec![encode(&raw), encode(&raw_null)], + expected_output: vec![ + MockUpdate::Delete(expected.clone()), + MockUpdate::Insert(expected.clone()), + MockUpdate::Delete(expected_null.clone()), + MockUpdate::Insert(expected_null.clone()), + ], + }; + + run_parser_test(vec![test]); +} + +/// A Debezium temporal type mapped to an incompatible SQL column must be +/// rejected during schema validation. +#[test] +fn test_debezium_temporal_type_mismatch() { + use super::schema::validate_struct_schema; + + let value_schema = r#"{ + "type": "record", + "name": "Temporal", + "connect.name": "test_namespace.Temporal", + "fields": [ + { "name": "id", "type": { "type": "long", "connect.name": "io.debezium.time.Timestamp" } } + ] + }"#; + + let hoisted = super::coercion::hoist_coercible_types(value_schema).unwrap(); + let schema = AvroSchema::parse_str(&hoisted).unwrap(); + let resolved = ResolvedSchema::try_from(&schema).unwrap(); + let refs = resolved + .get_names() + .iter() + .map(|(name, schema)| (name.clone(), (*schema).clone())) + .collect(); + + // The `id` column is BIGINT, but the Avro field is a Debezium timestamp. + let fields = vec![Field::new("id".into(), ColumnType::bigint(false))]; + let err = validate_struct_schema(&schema, &refs, &fields).unwrap_err(); + assert!( + err.contains("io.debezium.time.Timestamp") && err.contains("TIMESTAMP"), + "unexpected error message: {err}" + ); +} + +/// Raw wire representation of the same instant encoded four ways: as a +/// `ZonedTimestamp` (string) and a `MicroTimestamp` (long), each destined for +/// both a `TIMESTAMP` and a `TIMESTAMP WITH TIME ZONE` column. +#[derive(Debug, Clone)] +struct TsTargetsRaw { + z_ts: String, + z_tstz: String, + n_ts: i64, + n_tstz: i64, +} + +serialize_table_record!(TsTargetsRaw[4]{ + r#z_ts["z_ts"]: String, + r#z_tstz["z_tstz"]: String, + r#n_ts["n_ts"]: i64, + r#n_tstz["n_tstz"]: i64 +}); + +/// Decoded form of [`TsTargetsRaw`]. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TsTargets { + z_ts: Timestamp, + z_tstz: TimestampTz, + n_ts: Timestamp, + n_tstz: TimestampTz, +} + +deserialize_table_record!(TsTargets["TsTargets", Variant, 4] { + (r#z_ts, "z_ts", false, Timestamp, |_| None), + (r#z_tstz, "z_tstz", false, TimestampTz, |_| None), + (r#n_ts, "n_ts", false, Timestamp, |_| None), + (r#n_tstz, "n_tstz", false, TimestampTz, |_| None) +}); + +impl TsTargets { + fn value_avro_schema() -> &'static str { + r#"{ + "type": "record", + "name": "TsTargets", + "connect.name": "test_namespace.TsTargets", + "fields": [ + { "name": "z_ts", "type": { "type": "string", "connect.name": "io.debezium.time.ZonedTimestamp" } }, + { "name": "z_tstz", "type": { "type": "string", "connect.name": "io.debezium.time.ZonedTimestamp" } }, + { "name": "n_ts", "type": { "type": "long", "connect.name": "io.debezium.time.MicroTimestamp" } }, + { "name": "n_tstz", "type": { "type": "long", "connect.name": "io.debezium.time.MicroTimestamp" } } + ] + }"# + } + + fn relation_schema() -> Relation { + Relation { + name: SqlIdentifier::new("TsTargets", false), + fields: vec![ + Field::new("z_ts".into(), ColumnType::timestamp(false)), + Field::new("z_tstz".into(), ColumnType::timestamp_tz(false)), + Field::new("n_ts".into(), ColumnType::timestamp(false)), + Field::new("n_tstz".into(), ColumnType::timestamp_tz(false)), + ], + materialized: false, + properties: BTreeMap::new(), + primary_key: None, + } + } +} + +/// A Debezium timestamp type deserializes into either a `TIMESTAMP` or a +/// `TIMESTAMP WITH TIME ZONE` column. This covers both the string-encoded +/// `ZonedTimestamp` and a numeric type against both column types. +#[test] +fn test_debezium_timestamp_column_targets() { + let instant = DateTime::parse_from_rfc3339("2021-06-15T12:30:45.123Z") + .unwrap() + .with_timezone(&Utc); + let micros = instant.timestamp_micros(); + // Same instant expressed in a +02:00 offset. + let zoned = "2021-06-15T14:30:45.123+02:00".to_string(); + + let raw = TsTargetsRaw { + z_ts: zoned.clone(), + z_tstz: zoned.clone(), + n_ts: micros, + n_tstz: micros, + }; + let expected = TsTargets { + z_ts: Timestamp::from_microseconds(micros), + z_tstz: TimestampTz::from_microseconds(micros), + n_ts: Timestamp::from_microseconds(micros), + n_tstz: TimestampTz::from_microseconds(micros), + }; + + let envelope_str = debezium_avro_schema_str(TsTargets::value_avro_schema(), "TsTargets"); + let envelope_schema = AvroSchema::parse_str(&envelope_str).unwrap(); + let resolved = ResolvedSchema::try_from(&envelope_schema).unwrap(); + + let mut buffer = vec![0; 5]; + let serializer = AvroSchemaSerializer::new(&envelope_schema, resolved.get_names(), true); + let dbz_message = DebeziumMessage::new("u", Some(raw.clone()), Some(raw.clone())); + let val = dbz_message + .serialize_with_context(serializer, &avro_ser_config()) + .unwrap(); + let mut avro_record = to_avro_datum(&envelope_schema, val).unwrap(); + buffer.append(&mut avro_record); + + let test = TestCase { + relation_schema: TsTargets::relation_schema(), + config: AvroParserConfig { + update_format: AvroUpdateFormat::Debezium, + schema: Some(envelope_str), + skip_schema_id: false, + registry_config: Default::default(), + }, + input_batches: vec![(buffer, vec![])], + expected_output: vec![ + MockUpdate::Delete(expected.clone()), + MockUpdate::Insert(expected.clone()), + ], + }; + + run_parser_test(vec![test]); +} + +/// Raw wire representation of a Debezium `VariableScaleDecimal`: a record with +/// the scale and the big-endian two's-complement unscaled value. +#[derive(Debug, Clone)] +struct VariableScaleDecimalRaw { + scale: i32, + value: ByteArray, +} + +serialize_table_record!(VariableScaleDecimalRaw[2]{ + r#scale["scale"]: i32, + r#value["value"]: ByteArray +}); + +/// Raw wire representation of a row with variable-scale decimal columns, one of +/// them nullable and encoded via a by-name reference to the same record type. +#[derive(Debug, Clone)] +struct TestDecimalRaw { + id: i64, + amount: VariableScaleDecimalRaw, + amount_opt: Option, +} + +serialize_table_record!(TestDecimalRaw[3]{ + r#id["id"]: i64, + r#amount["amount"]: VariableScaleDecimalRaw, + r#amount_opt["amount_opt"]: Option +}); + +/// Decoded representation of [`TestDecimalRaw`]: every variable-scale decimal +/// parsed into a Feldera `DECIMAL` column. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct TestDecimal { + id: i64, + amount: SqlDecimal<10, 3>, + amount_opt: Option>, +} + +deserialize_table_record!(TestDecimal["Dec", Variant, 3] { + (r#id, "id", false, i64, |_| None), + (r#amount, "amount", false, SqlDecimal<10, 3>, |_| None), + (r#amount_opt, "amount_opt", true, Option>, |_| Some(None)) +}); + +impl TestDecimal { + /// Avro value schema with two `VariableScaleDecimal` columns; `amount_opt` + /// references the record type by name, exercising reference resolution. + fn value_avro_schema() -> &'static str { + r#"{ + "type": "record", + "name": "Dec", + "connect.name": "test_namespace.Dec", + "fields": [ + { "name": "id", "type": "long" }, + { "name": "amount", "type": { + "type": "record", + "name": "VariableScaleDecimal", + "namespace": "io.debezium.data", + "fields": [ + { "name": "scale", "type": "int" }, + { "name": "value", "type": "bytes" } + ], + "connect.version": 1, + "connect.name": "io.debezium.data.VariableScaleDecimal" + }}, + { "name": "amount_opt", "type": ["null", "io.debezium.data.VariableScaleDecimal"], "default": null } + ] + }"# + } + + fn relation_schema() -> Relation { + Relation { + name: SqlIdentifier::new("Dec", false), + fields: vec![ + Field::new("id".into(), ColumnType::bigint(false)), + Field::new("amount".into(), ColumnType::decimal(10, 3, false)), + Field::new("amount_opt".into(), ColumnType::decimal(10, 3, true)), + ], + materialized: false, + properties: BTreeMap::new(), + primary_key: None, + } + } +} + +/// Parse Debezium `io.debezium.data.VariableScaleDecimal` records into Feldera +/// `DECIMAL` columns, including a nullable column and a by-name type reference. +#[test] +fn test_debezium_variable_scale_decimal() { + let raw_dec = |scale: i32, unscaled: i64| VariableScaleDecimalRaw { + scale, + value: ByteArray::new(&BigInt::from(unscaled).to_signed_bytes_be()), + }; + + // The wire scale differs from the column scale (3), which is the whole point + // of the variable-scale type. + let row1 = TestDecimalRaw { + id: 1, + amount: raw_dec(2, 12345), // 123.45 + amount_opt: Some(raw_dec(3, -6789)), // -6.789 + }; + let row2 = TestDecimalRaw { + id: 2, + amount: raw_dec(0, 42), // 42 + amount_opt: None, + }; + + let expected1 = TestDecimal { + id: 1, + amount: SqlDecimal::<10, 3>::new(12345, 2).unwrap(), + amount_opt: Some(SqlDecimal::<10, 3>::new(-6789, 3).unwrap()), + }; + let expected2 = TestDecimal { + id: 2, + amount: SqlDecimal::<10, 3>::new(42, 0).unwrap(), + amount_opt: None, + }; + + let envelope_str = debezium_avro_schema_str(TestDecimal::value_avro_schema(), "Dec"); + let envelope_schema = AvroSchema::parse_str(&envelope_str).unwrap(); + let resolved = ResolvedSchema::try_from(&envelope_schema).unwrap(); + + let encode = |row: &TestDecimalRaw| { + let mut buffer = vec![0; 5]; + let serializer = AvroSchemaSerializer::new(&envelope_schema, resolved.get_names(), true); + let dbz_message = DebeziumMessage::new("u", Some(row.clone()), Some(row.clone())); + let val = dbz_message + .serialize_with_context(serializer, &avro_ser_config()) + .unwrap(); + let mut avro_record = to_avro_datum(&envelope_schema, val).unwrap(); + buffer.append(&mut avro_record); + (buffer, vec![]) + }; + + let test = TestCase { + relation_schema: TestDecimal::relation_schema(), + config: AvroParserConfig { + update_format: AvroUpdateFormat::Debezium, + schema: Some(envelope_str), + skip_schema_id: false, + registry_config: Default::default(), + }, + input_batches: vec![encode(&row1), encode(&row2)], + expected_output: vec![ + MockUpdate::Delete(expected1.clone()), + MockUpdate::Insert(expected1.clone()), + MockUpdate::Delete(expected2.clone()), + MockUpdate::Insert(expected2.clone()), + ], + }; + + run_parser_test(vec![test]); +} + /// Type used to serialize different integer types as Avro `int`. #[derive( Debug, diff --git a/crates/feldera-types/src/format/avro.rs b/crates/feldera-types/src/format/avro.rs index 7f2b6b046f1..0ed15ea5767 100644 --- a/crates/feldera-types/src/format/avro.rs +++ b/crates/feldera-types/src/format/avro.rs @@ -25,6 +25,36 @@ pub enum AvroUpdateFormat { Raw, /// Debezium data change event format. + /// + /// ### Temporal types + /// + /// Debezium encodes temporal columns with a `connect.name` annotation + /// rather than a native Avro logical type. The input connector recognizes + /// these annotations and converts each value into the matching Feldera + /// column type: + /// + /// | `connect.name` | wire encoding | Feldera column | + /// |-----------------------------------------|-----------------------------------|----------------| + /// | `io.debezium.time.Date` | `int` days since epoch | `DATE` | + /// | `io.debezium.time.Time` | `int` milliseconds since midnight | `TIME` | + /// | `io.debezium.time.MicroTime` | `long` microseconds since midnight| `TIME` | + /// | `io.debezium.time.NanoTime` | `long` nanoseconds since midnight | `TIME` | + /// | `io.debezium.time.ZonedTime` | ISO-8601 `string` | `TIME` | + /// | `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` | + /// + /// Any timestamp type populates either a `TIMESTAMP` or a `TIMESTAMP WITH + /// TIME ZONE` column; the connector stores the same instant (microseconds + /// since the Unix epoch) in both cases. + /// + /// The `org.apache.kafka.connect.data.{Date,Time,Timestamp}` types emitted + /// with `time.precision.mode=connect` are also supported. + /// + /// `io.debezium.data.VariableScaleDecimal` (used for `NUMERIC`/`DECIMAL` + /// columns without a fixed scale, encoded as a `{scale, value}` record) is + /// parsed into a `DECIMAL` column. #[serde(rename = "debezium")] Debezium, diff --git a/docs.feldera.com/docs/formats/avro.md b/docs.feldera.com/docs/formats/avro.md index 18386d10f71..cc03c15e73a 100644 --- a/docs.feldera.com/docs/formats/avro.md +++ b/docs.feldera.com/docs/formats/avro.md @@ -96,6 +96,40 @@ A SQL column and a field in the Avro schema are compatible if the following cond | `VARIANT` | `string` | values of type `VARIANT` are deserialized from JSON-encoded strings (see [`VARIANT` documetation](/sql/json)) | | user-defined types | `record` | Avro record schema must match SQL user-defined type definition according to the same schema compatibility rules as for SQL tables| +### Debezium logical types + +When `update_format` is set to `debezium`, the connector additionally recognizes +Debezium's temporal and decimal +[semantic types](https://debezium.io/documentation/reference/stable/connectors/postgresql.html#postgresql-data-types). +Debezium encodes these as plain Avro types annotated with a `connect.name` +attribute rather than as native Avro logical types. The connector converts each +value into the matching SQL column type: + +| `connect.name` | Avro encoding | SQL column | +|-----------------------------------------|-------------------------------------------------------------|-------------| +| `io.debezium.time.Date` | `int` days since epoch | `DATE` | +| `io.debezium.time.Time` | `int` milliseconds since midnight | `TIME` | +| `io.debezium.time.MicroTime` | `long` microseconds since midnight | `TIME` | +| `io.debezium.time.NanoTime` | `long` nanoseconds since midnight | `TIME` | +| `io.debezium.time.ZonedTime` | ISO-8601 `string` (e.g. `10:15:30+01:00`) | `TIME` | +| `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` (e.g. `2011-12-03T10:15:30.030431+01:00`) | `TIMESTAMP` | +| `io.debezium.data.VariableScaleDecimal` | `{scale, value}` record | `DECIMAL` | + +Notes: + +* Any timestamp type can be deserialized into either a `TIMESTAMP` or a + `TIMESTAMP WITH TIME ZONE` column; the connector stores the same instant in + both cases. Zoned values are normalized to UTC. +* The equivalent `org.apache.kafka.connect.data.Date`, `.Time`, and `.Timestamp` + types, emitted when Debezium's `time.precision.mode` is set to `connect`, are + also supported. +* `io.debezium.data.VariableScaleDecimal` represents decimals whose scale is not + fixed in the schema (for example, an unconstrained PostgreSQL `NUMERIC`). The + wire scale may differ from the SQL column's scale. + ### Configuration The following properties can be used to configure the Avro parser. All of these properties are optional. However,