diff --git a/crates/adapters/src/format/avro/deserializer.rs b/crates/adapters/src/format/avro/deserializer.rs index df8dc257d4e..870f1df897a 100644 --- a/crates/adapters/src/format/avro/deserializer.rs +++ b/crates/adapters/src/format/avro/deserializer.rs @@ -501,7 +501,7 @@ impl<'de> de::Deserializer<'de> for &'_ Deserializer<'de> { where V: Visitor<'de>, { - self.deserialize_any(visitor) + visitor.visit_unit() } fn is_human_readable(&self) -> bool { @@ -637,3 +637,27 @@ pub fn from_avro_value<'de, D: DeserializeWithContext<'de, SqlSerdeConfig, AUX>, D::deserialize_with_context_aux(deserializer, avro_de_config(), metadata) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::format::avro::{coercion::Coercion, debezium::DebeziumTimeType}; + + #[test] + fn ignore_nullable_coerced_null() { + // Extra Avro record fields are consumed as `IgnoredAny`. A nullable + // Debezium timestamp retains its coercion even though the target table + // does not contain the field. + let input = Value::Union(0, Box::new(Value::Null)); + let schema = Schema::parse_str(r#"["null", "string"]"#).unwrap(); + let refs = AvroSchemaRefs::new(); + let deserializer = Deserializer::with_coercion( + &input, + &schema, + &refs, + Some(Coercion::DebeziumTime(DebeziumTimeType::ZonedTimestamp)), + ); + + ::deserialize(&deserializer).unwrap(); + } +} diff --git a/crates/adapters/src/format/avro/test.rs b/crates/adapters/src/format/avro/test.rs index d0dbfb32e61..cc79b0d64ee 100644 --- a/crates/adapters/src/format/avro/test.rs +++ b/crates/adapters/src/format/avro/test.rs @@ -401,6 +401,97 @@ fn test_debezium_avro_parser() { run_parser_test(vec![test_case]) } +/// Source row containing a Debezium logical field that is not declared in the +/// destination table. +#[derive(Debug, Clone)] +struct IgnoredCoercedFieldSource { + id: i64, + timestamp: Option, +} + +serialize_table_record!(IgnoredCoercedFieldSource[2]{ + r#id["id"]: i64, + r#timestamp["timestamp"]: Option +}); + +/// Destination row intentionally omitting `timestamp`. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +struct IgnoredCoercedFieldRow { + id: i64, +} + +deserialize_table_record!(IgnoredCoercedFieldRow["IgnoredCoercedField", Variant, 1] { + (r#id, "id", false, i64, |_| None) +}); + +/// Process a complete Debezium update whose omitted `ZonedTimestamp` field is +/// NULL in `before` and populated in `after`. Neither value is relevant to the +/// destination table, so both records must be accepted without coercion. +#[test] +fn test_debezium_ignored_coerced_field() { + let value_schema = r#"{ + "type": "record", + "name": "IgnoredCoercedField", + "connect.name": "test_namespace.IgnoredCoercedField", + "fields": [ + { "name": "id", "type": "long" }, + { + "name": "timestamp", + "type": [ + "null", + { + "type": "string", + "connect.name": "io.debezium.time.ZonedTimestamp" + } + ], + "default": null + } + ] + }"#; + let envelope_str = debezium_avro_schema_str(value_schema, "IgnoredCoercedField"); + let envelope_schema = AvroSchema::parse_str(&envelope_str).unwrap(); + let resolved = ResolvedSchema::try_from(&envelope_schema).unwrap(); + let serializer = AvroSchemaSerializer::new(&envelope_schema, resolved.get_names(), true); + + let message = DebeziumMessage::new( + "u", + Some(IgnoredCoercedFieldSource { + id: 1, + timestamp: None, + }), + Some(IgnoredCoercedFieldSource { + id: 2, + timestamp: Some("2021-06-15T14:30:45.123+02:00".to_string()), + }), + ); + let value = message + .serialize_with_context(serializer, &avro_ser_config()) + .unwrap(); + + let test = TestCase { + relation_schema: Relation { + name: SqlIdentifier::new("IgnoredCoercedField", false), + fields: vec![Field::new("id".into(), ColumnType::bigint(false))], + materialized: false, + properties: BTreeMap::new(), + primary_key: None, + }, + config: AvroParserConfig { + update_format: AvroUpdateFormat::Debezium, + schema: Some(envelope_str), + skip_schema_id: false, + registry_config: Default::default(), + }, + input_batches: vec![(serialize_value(value, &envelope_schema), vec![])], + expected_output: vec![ + MockUpdate::Delete(IgnoredCoercedFieldRow { id: 1 }), + MockUpdate::Insert(IgnoredCoercedFieldRow { id: 2 }), + ], + }; + + run_parser_test(vec![test]); +} + /// SQL table can have nullable columns that are not in the Avro schema. #[test] fn test_extra_columns() {