risingwave_connector_codec/decoder/avro/
mod.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15mod schema;
16
17use apache_avro::Schema;
18use apache_avro::schema::{DecimalSchema, NamesRef, UnionSchema};
19use apache_avro::types::{Value, ValueKind};
20use chrono::Datelike;
21use itertools::Itertools;
22use num_bigint::BigInt;
23use risingwave_common::array::{ListValue, StructValue};
24use risingwave_common::types::{
25    DataType, Date, DatumCow, Interval, JsonbVal, MapValue, ScalarImpl, Time, Timestamp,
26    Timestamptz, ToOwnedDatum,
27};
28
29pub use self::schema::{MapHandling, ResolvedAvroSchema, avro_schema_to_fields};
30use super::utils::scaled_bigint_to_rust_decimal;
31use super::{Access, AccessError, AccessResult, bail_uncategorized, uncategorized};
32use crate::decoder::avro::schema::avro_schema_to_struct_field_name;
33
34#[derive(Clone)]
35/// Options for parsing an `AvroValue` into Datum, with a root avro schema.
36pub struct AvroParseOptions<'a> {
37    /// The avro schema at root level
38    root_schema: &'a Schema,
39    /// The immutable "global" context during recursive parsing
40    inner: AvroParseOptionsInner<'a>,
41}
42
43#[derive(Clone)]
44/// Options for parsing an `AvroValue` into Datum, with names resolved from root schema.
45struct AvroParseOptionsInner<'a> {
46    /// Mapping from type names to actual schema
47    refs: NamesRef<'a>,
48    /// Strict Mode
49    /// If strict mode is disabled, an int64 can be parsed from an `AvroInt` (int32) value.
50    relax_numeric: bool,
51}
52
53impl<'a> AvroParseOptions<'a> {
54    pub fn create(root_schema: &'a Schema) -> Self {
55        let resolved = apache_avro::schema::ResolvedSchema::try_from(root_schema)
56            .expect("avro schema is self contained");
57        Self {
58            root_schema,
59            inner: AvroParseOptionsInner {
60                refs: resolved.get_names().clone(),
61                relax_numeric: true,
62            },
63        }
64    }
65}
66
67impl<'a> AvroParseOptionsInner<'a> {
68    fn lookup_ref(&self, schema: &'a Schema) -> &'a Schema {
69        match schema {
70            Schema::Ref { name } => self.refs[name],
71            _ => schema,
72        }
73    }
74
75    /// Parse an avro value into expected type.
76    ///
77    /// 3 kinds of type info are used to parsing:
78    /// - `type_expected`. The type that we expect the value is.
79    /// - value type. The type info together with the value argument.
80    /// - schema. The `AvroSchema` provided in option.
81    ///
82    /// Cases: (FIXME: Is this precise?)
83    /// - If both `type_expected` and schema are provided, it will check both strictly.
84    /// - If only `type_expected` is provided, it will try to match the value type
85    ///   and the `type_expected`, converting the value if possible.
86    /// - If only value is provided (without schema and `type_expected`),
87    ///   the `DataType` will be inferred.
88    fn convert_to_datum<'b>(
89        &self,
90        unresolved_schema: &'a Schema,
91        value: &'b Value,
92        type_expected: &DataType,
93    ) -> AccessResult<DatumCow<'b>>
94    where
95        'b: 'a,
96    {
97        let create_error = || AccessError::TypeError {
98            expected: format!("{:?}", type_expected),
99            got: format!("{:?}", value),
100            value: String::new(),
101        };
102
103        macro_rules! borrowed {
104            ($v:expr) => {
105                return Ok(DatumCow::Borrowed(Some($v.into())))
106            };
107        }
108
109        let v: ScalarImpl = match (type_expected, value) {
110            (_, Value::Null) => return Ok(DatumCow::NULL),
111            // ---- Union (with >=2 non null variants), and nullable Union ([null, record]) -----
112            (DataType::Struct(struct_type_info), Value::Union(variant, v)) => {
113                let Schema::Union(u) = self.lookup_ref(unresolved_schema) else {
114                    // XXX: Is this branch actually unreachable? (if self.schema is correctly used)
115                    return Err(create_error());
116                };
117
118                if let Some(inner) = get_nullable_union_inner(u) {
119                    // nullable Union ([null, record])
120                    return self.convert_to_datum(inner, v, type_expected);
121                }
122                let variant_schema = &u.variants()[*variant as usize];
123
124                if matches!(variant_schema, &Schema::Null) {
125                    return Ok(DatumCow::NULL);
126                }
127
128                // Here we compare the field name, instead of using the variant idx to find the field idx.
129                // The latter approach might also work, but might be more error-prone.
130                // We will need to get the index of the "null" variant, and then re-map the variant index to the field index.
131                // XXX: probably we can unwrap here (if self.schema is correctly used)
132                let expected_field_name = avro_schema_to_struct_field_name(variant_schema)?;
133
134                let mut fields = Vec::with_capacity(struct_type_info.len());
135                for (field_name, field_type) in struct_type_info.iter() {
136                    if field_name == expected_field_name {
137                        let datum = self
138                            .convert_to_datum(variant_schema, v, field_type)?
139                            .to_owned_datum();
140
141                        fields.push(datum)
142                    } else {
143                        fields.push(None)
144                    }
145                }
146                StructValue::new(fields).into()
147            }
148            // nullable Union ([null, T])
149            (_, Value::Union(_, v)) => {
150                let Schema::Union(u) = self.lookup_ref(unresolved_schema) else {
151                    return Err(create_error());
152                };
153                let Some(schema) = get_nullable_union_inner(u) else {
154                    return Err(create_error());
155                };
156                return self.convert_to_datum(schema, v, type_expected);
157            }
158            // ---- Boolean -----
159            (DataType::Boolean, Value::Boolean(b)) => (*b).into(),
160            // ---- Int16 -----
161            (DataType::Int16, Value::Int(i)) if self.relax_numeric => (*i as i16).into(),
162            (DataType::Int16, Value::Long(i)) if self.relax_numeric => (*i as i16).into(),
163
164            // ---- Int32 -----
165            (DataType::Int32, Value::Int(i)) => (*i).into(),
166            (DataType::Int32, Value::Long(i)) if self.relax_numeric => (*i as i32).into(),
167            // ---- Int64 -----
168            (DataType::Int64, Value::Long(i)) => (*i).into(),
169            (DataType::Int64, Value::Int(i)) if self.relax_numeric => (*i as i64).into(),
170            // ---- Float32 -----
171            (DataType::Float32, Value::Float(i)) => (*i).into(),
172            (DataType::Float32, Value::Double(i)) => (*i as f32).into(),
173            // ---- Float64 -----
174            (DataType::Float64, Value::Double(i)) => (*i).into(),
175            (DataType::Float64, Value::Float(i)) => (*i as f64).into(),
176            // ---- Decimal -----
177            (DataType::Decimal, Value::Decimal(avro_decimal)) => {
178                let (_precision, scale) = match self.lookup_ref(unresolved_schema) {
179                    Schema::Decimal(DecimalSchema {
180                        precision, scale, ..
181                    }) => (*precision, *scale),
182                    _ => Err(create_error())?,
183                };
184                let decimal = scaled_bigint_to_rust_decimal(avro_decimal.clone().into(), scale)
185                    .map_err(|_| create_error())?;
186                ScalarImpl::Decimal(risingwave_common::types::Decimal::Normalized(decimal))
187            }
188            (DataType::Decimal, Value::Record(fields)) => {
189                // VariableScaleDecimal has fixed fields, scale(int) and value(bytes)
190                let find_in_records = |field_name: &str| {
191                    fields
192                        .iter()
193                        .find(|field| field.0 == field_name)
194                        .map(|field| &field.1)
195                        .ok_or_else(|| {
196                            uncategorized!("`{field_name}` field not found in VariableScaleDecimal")
197                        })
198                };
199                let scale = match find_in_records("scale")? {
200                    Value::Int(scale) => *scale,
201                    avro_value => bail_uncategorized!(
202                        "scale field in VariableScaleDecimal is not int, got {:?}",
203                        avro_value
204                    ),
205                };
206
207                let value: BigInt = match find_in_records("value")? {
208                    Value::Bytes(bytes) => BigInt::from_signed_bytes_be(bytes),
209                    avro_value => bail_uncategorized!(
210                        "value field in VariableScaleDecimal is not bytes, got {:?}",
211                        avro_value
212                    ),
213                };
214
215                let decimal = scaled_bigint_to_rust_decimal(value, scale as _)?;
216                ScalarImpl::Decimal(risingwave_common::types::Decimal::Normalized(decimal))
217            }
218            // ---- Time -----
219            (DataType::Time, Value::TimeMillis(ms)) => Time::with_milli(*ms as u32)
220                .map_err(|_| create_error())?
221                .into(),
222            (DataType::Time, Value::TimeMicros(us)) => Time::with_micro(*us as u64)
223                .map_err(|_| create_error())?
224                .into(),
225            // ---- Date -----
226            (DataType::Date, Value::Date(days)) => {
227                Date::with_days_since_ce(days + unix_epoch_days())
228                    .map_err(|_| create_error())?
229                    .into()
230            }
231            // ---- Varchar -----
232            (DataType::Varchar, Value::Enum(_, symbol)) => borrowed!(symbol.as_str()),
233            (DataType::Varchar, Value::String(s)) => borrowed!(s.as_str()),
234            // ---- Timestamp -----
235            (DataType::Timestamp, Value::LocalTimestampMillis(ms)) => Timestamp::with_millis(*ms)
236                .map_err(|_| create_error())?
237                .into(),
238            (DataType::Timestamp, Value::LocalTimestampMicros(us)) => Timestamp::with_micros(*us)
239                .map_err(|_| create_error())?
240                .into(),
241
242            // ---- TimestampTz -----
243            (DataType::Timestamptz, Value::TimestampMillis(ms)) => Timestamptz::from_millis(*ms)
244                .ok_or_else(|| {
245                    uncategorized!("timestamptz with milliseconds {ms} * 1000 is out of range")
246                })?
247                .into(),
248            (DataType::Timestamptz, Value::TimestampMicros(us)) => Timestamptz::from_micros(*us)
249                .ok_or_else(|| {
250                    uncategorized!("timestamptz with microseconds {us} is out of range")
251                })?
252                .into(),
253
254            // ---- Interval -----
255            (DataType::Interval, Value::Duration(duration)) => {
256                let months = u32::from(duration.months()) as i32;
257                let days = u32::from(duration.days()) as i32;
258                let usecs = (u32::from(duration.millis()) as i64) * 1000; // never overflows
259                ScalarImpl::Interval(Interval::from_month_day_usec(months, days, usecs))
260            }
261            // ---- Struct -----
262            (DataType::Struct(struct_type_info), Value::Record(descs)) => StructValue::new({
263                let Schema::Record(record_schema) = self.lookup_ref(unresolved_schema) else {
264                    return Err(create_error());
265                };
266                struct_type_info
267                    .iter()
268                    .map(|(field_name, field_type)| {
269                        if let Some(idx) = record_schema.lookup.get(field_name) {
270                            let value = &descs[*idx].1;
271                            let schema = &record_schema.fields[*idx].schema;
272                            Ok(self
273                                .convert_to_datum(schema, value, field_type)?
274                                .to_owned_datum())
275                        } else {
276                            Ok(None)
277                        }
278                    })
279                    .collect::<Result<_, AccessError>>()?
280            })
281            .into(),
282            // ---- List -----
283            (DataType::List(list_type), Value::Array(array)) => ListValue::new({
284                let Schema::Array(array_schema) = self.lookup_ref(unresolved_schema) else {
285                    return Err(create_error());
286                };
287                let schema = &array_schema.items;
288                let elem_type = list_type.elem();
289                let mut builder = elem_type.create_array_builder(array.len());
290                for v in array {
291                    let value = self.convert_to_datum(schema, v, elem_type)?;
292                    builder.append(value);
293                }
294                builder.finish()
295            })
296            .into(),
297            // ---- Bytea -----
298            (DataType::Bytea, Value::Bytes(value)) => borrowed!(value.as_slice()),
299            // ---- Jsonb -----
300            (DataType::Jsonb, v @ Value::Map(_)) => {
301                let mut builder = jsonbb::Builder::default();
302                avro_to_jsonb(v, &mut builder)?;
303                let jsonb = builder.finish();
304                debug_assert!(jsonb.as_ref().is_object());
305                JsonbVal::from(jsonb).into()
306            }
307            (DataType::Varchar, Value::Uuid(uuid)) => {
308                uuid.as_hyphenated().to_string().into_boxed_str().into()
309            }
310            (DataType::Map(map_type), Value::Map(map)) => {
311                let Schema::Map(map_schema) = self.lookup_ref(unresolved_schema) else {
312                    return Err(create_error());
313                };
314                let schema = &map_schema.types;
315                let mut builder = map_type
316                    .clone()
317                    .into_struct()
318                    .create_array_builder(map.len());
319                // Since the map is HashMap, we can ensure
320                // key is non-null and unique, keys and values have the same length.
321
322                // NOTE: HashMap's iter order is non-deterministic, but MapValue's
323                // order matters. We sort by key here to have deterministic order
324                // in tests. We might consider removing this, or make all MapValue sorted
325                // in the future.
326                for (k, v) in map.iter().sorted_by_key(|(k, _v)| *k) {
327                    let value_datum = self
328                        .convert_to_datum(schema, v, map_type.value())?
329                        .to_owned_datum();
330                    builder.append(
331                        StructValue::new(vec![Some(k.as_str().into()), value_datum])
332                            .to_owned_datum(),
333                    );
334                }
335                let list = ListValue::new(builder.finish());
336                MapValue::from_entries(list).into()
337            }
338
339            (_expected, _got) => Err(create_error())?,
340        };
341        Ok(DatumCow::Owned(Some(v)))
342    }
343}
344
345pub struct AvroAccess<'a> {
346    value: &'a Value,
347    options: AvroParseOptions<'a>,
348}
349
350impl<'a> AvroAccess<'a> {
351    pub fn new(root_value: &'a Value, options: AvroParseOptions<'a>) -> Self {
352        Self {
353            value: root_value,
354            options,
355        }
356    }
357}
358
359impl Access for AvroAccess<'_> {
360    fn access<'a>(&'a self, path: &[&str], type_expected: &DataType) -> AccessResult<DatumCow<'a>> {
361        let mut value = self.value;
362        let mut unresolved_schema = self.options.root_schema;
363
364        debug_assert!(
365            path.len() == 1
366                || (path.len() == 2 && matches!(path[0], "before" | "after" | "source")),
367            "unexpected path access: {:?}",
368            path
369        );
370        let mut i = 0;
371        while i < path.len() {
372            let key = path[i];
373            let create_error = || AccessError::Undefined {
374                name: key.to_owned(),
375                path: path.iter().take(i).join("."),
376            };
377            match value {
378                Value::Union(_, v) => {
379                    // The debezium "before" field is a nullable union.
380                    // "fields": [
381                    // {
382                    //     "name": "before",
383                    //     "type": [
384                    //         "null",
385                    //         {
386                    //             "type": "record",
387                    //             "name": "Value",
388                    //             "fields": [...],
389                    //         }
390                    //     ],
391                    //     "default": null
392                    // },
393                    // {
394                    //     "name": "after",
395                    //     "type": [
396                    //         "null",
397                    //         "Value"
398                    //     ],
399                    //     "default": null
400                    // },
401                    // ...]
402                    value = v;
403                    let Schema::Union(u) = self.options.inner.lookup_ref(unresolved_schema) else {
404                        return Err(create_error());
405                    };
406                    let Some(schema) = get_nullable_union_inner(u) else {
407                        return Err(create_error());
408                    };
409                    unresolved_schema = schema;
410                    continue;
411                }
412                Value::Record(fields) => {
413                    let Schema::Record(record_schema) =
414                        self.options.inner.lookup_ref(unresolved_schema)
415                    else {
416                        return Err(create_error());
417                    };
418                    if let Some(idx) = record_schema.lookup.get(key) {
419                        value = &fields[*idx].1;
420                        unresolved_schema = &record_schema.fields[*idx].schema;
421                        i += 1;
422                        continue;
423                    }
424                }
425                _ => (),
426            }
427            Err(create_error())?;
428        }
429
430        self.options
431            .inner
432            .convert_to_datum(unresolved_schema, value, type_expected)
433    }
434}
435
436/// If the union schema is `[null, T]` or `[T, null]`, returns `Some(T)`; otherwise returns `None`.
437pub fn get_nullable_union_inner(union_schema: &UnionSchema) -> Option<&'_ Schema> {
438    let variants = union_schema.variants();
439    // Note: `[null, null] is invalid`, we don't need to worry about that.
440    if variants.len() == 2 && variants.contains(&Schema::Null) {
441        let inner_schema = variants
442            .iter()
443            .find(|s| !matches!(s, &&Schema::Null))
444            .unwrap();
445        Some(inner_schema)
446    } else {
447        None
448    }
449}
450
451pub(crate) fn unix_epoch_days() -> i32 {
452    Date::from_ymd_uncheck(1970, 1, 1).0.num_days_from_ce()
453}
454
455pub(crate) fn avro_to_jsonb(avro: &Value, builder: &mut jsonbb::Builder) -> AccessResult<()> {
456    match avro {
457        Value::Null => builder.add_null(),
458        Value::Boolean(b) => builder.add_bool(*b),
459        Value::Int(i) => builder.add_i64(*i as i64),
460        Value::String(s) => builder.add_string(s),
461        Value::Map(m) => {
462            builder.begin_object();
463            for (k, v) in m {
464                builder.add_string(k);
465                avro_to_jsonb(v, builder)?;
466            }
467            builder.end_object()
468        }
469        // same representation as map
470        Value::Record(r) => {
471            builder.begin_object();
472            for (k, v) in r {
473                builder.add_string(k);
474                avro_to_jsonb(v, builder)?;
475            }
476            builder.end_object()
477        }
478        Value::Array(a) => {
479            builder.begin_array();
480            for v in a {
481                avro_to_jsonb(v, builder)?;
482            }
483            builder.end_array()
484        }
485
486        // TODO: figure out where the following encoding is reasonable before enabling them.
487        // See discussions: https://github.com/risingwavelabs/risingwave/pull/16948
488
489        // jsonbb supports int64, but JSON spec does not allow it. How should we handle it?
490        // BTW, protobuf canonical JSON converts int64 to string.
491        // Value::Long(l) => builder.add_i64(*l),
492        // Value::Float(f) => {
493        //     if f.is_nan() || f.is_infinite() {
494        //         // XXX: pad null or return err here?
495        //         builder.add_null()
496        //     } else {
497        //         builder.add_f64(*f as f64)
498        //     }
499        // }
500        // Value::Double(f) => {
501        //     if f.is_nan() || f.is_infinite() {
502        //         // XXX: pad null or return err here?
503        //         builder.add_null()
504        //     } else {
505        //         builder.add_f64(*f)
506        //     }
507        // }
508        // // XXX: What encoding to use?
509        // // ToText is \x plus hex string.
510        // Value::Bytes(b) => builder.add_string(&ToText::to_text(&b.as_slice())),
511        // Value::Enum(_, symbol) => {
512        //     builder.add_string(&symbol);
513        // }
514        // Value::Uuid(id) => builder.add_string(&id.as_hyphenated().to_string()),
515        // // For Union, one concern is that the avro union is tagged (like rust enum) but json union is untagged (like c union).
516        // // When the union consists of multiple records, it is possible to distinguish which variant is active in avro, but in json they will all become jsonb objects and indistinguishable.
517        // Value::Union(_, v) => avro_to_jsonb(v, builder)?
518        // XXX: pad null or return err here?
519        v @ (Value::Long(_)
520        | Value::Float(_)
521        | Value::Double(_)
522        | Value::Bytes(_)
523        | Value::Enum(_, _)
524        | Value::Fixed(_, _)
525        | Value::Date(_)
526        | Value::Decimal(_)
527        | Value::BigDecimal(_)
528        | Value::TimeMillis(_)
529        | Value::TimeMicros(_)
530        | Value::TimestampMillis(_)
531        | Value::TimestampMicros(_)
532        | Value::TimestampNanos(_)
533        | Value::LocalTimestampMillis(_)
534        | Value::LocalTimestampMicros(_)
535        | Value::LocalTimestampNanos(_)
536        | Value::Duration(_)
537        | Value::Uuid(_)
538        | Value::Union(_, _)) => {
539            bail_uncategorized!(
540                "unimplemented conversion from avro to jsonb: {:?}",
541                ValueKind::from(v)
542            )
543        }
544    }
545    Ok(())
546}
547
548#[cfg(test)]
549mod tests {
550    use std::str::FromStr;
551
552    use apache_avro::{Decimal as AvroDecimal, from_avro_datum};
553    use expect_test::expect;
554    use risingwave_common::types::{Datum, Decimal};
555
556    use super::*;
557
558    /// Test the behavior of the Rust Avro lib for handling union with logical type.
559    #[test]
560    fn test_avro_lib_union() {
561        // duplicate types
562        let s = Schema::parse_str(r#"["null", "null"]"#);
563        expect![[r#"
564            Err(
565                Error {
566                    details: Unions cannot contain duplicate types,
567                },
568            )
569        "#]]
570        .assert_debug_eq(&s);
571        let s = Schema::parse_str(r#"["int", "int"]"#);
572        expect![[r#"
573            Err(
574                Error {
575                    details: Unions cannot contain duplicate types,
576                },
577            )
578        "#]]
579        .assert_debug_eq(&s);
580        // multiple map/array are considered as the same type, regardless of the element type!
581        let s = Schema::parse_str(
582            r#"[
583"null",
584{
585    "type": "map",
586    "values" : "long",
587    "default": {}
588},
589{
590    "type": "map",
591    "values" : "int",
592    "default": {}
593}
594]
595"#,
596        );
597        expect![[r#"
598            Err(
599                Error {
600                    details: Unions cannot contain duplicate types,
601                },
602            )
603        "#]]
604        .assert_debug_eq(&s);
605        let s = Schema::parse_str(
606            r#"[
607"null",
608{
609    "type": "array",
610    "items" : "long",
611    "default": {}
612},
613{
614    "type": "array",
615    "items" : "int",
616    "default": {}
617}
618]
619"#,
620        );
621        expect![[r#"
622            Err(
623                Error {
624                    details: Unions cannot contain duplicate types,
625                },
626            )
627        "#]]
628        .assert_debug_eq(&s);
629        // multiple named types
630        let s = Schema::parse_str(
631            r#"[
632"null",
633{"type":"fixed","name":"a","size":16},
634{"type":"fixed","name":"b","size":32}
635]
636"#,
637        );
638        expect![[r#"
639            Ok(
640                Union(
641                    UnionSchema {
642                        schemas: [
643                            Null,
644                            Fixed(
645                                FixedSchema {
646                                    name: Name {
647                                        name: "a",
648                                        namespace: None,
649                                    },
650                                    aliases: None,
651                                    doc: None,
652                                    size: 16,
653                                    default: None,
654                                    attributes: {},
655                                },
656                            ),
657                            Fixed(
658                                FixedSchema {
659                                    name: Name {
660                                        name: "b",
661                                        namespace: None,
662                                    },
663                                    aliases: None,
664                                    doc: None,
665                                    size: 32,
666                                    default: None,
667                                    attributes: {},
668                                },
669                            ),
670                        ],
671                        variant_index: {
672                            Null: 0,
673                        },
674                    },
675                ),
676            )
677        "#]]
678        .assert_debug_eq(&s);
679
680        // union in union
681        let s = Schema::parse_str(r#"["int", ["null", "int"]]"#);
682        expect![[r#"
683            Err(
684                Error {
685                    details: Unions may not directly contain a union,
686                },
687            )
688        "#]]
689        .assert_debug_eq(&s);
690
691        // logical type
692        let s = Schema::parse_str(r#"["null", {"type":"string","logicalType":"uuid"}]"#).unwrap();
693        expect![[r#"
694            Union(
695                UnionSchema {
696                    schemas: [
697                        Null,
698                        Uuid,
699                    ],
700                    variant_index: {
701                        Null: 0,
702                        Uuid: 1,
703                    },
704                },
705            )
706        "#]]
707        .assert_debug_eq(&s);
708        // Note: Java Avro lib rejects this (logical type unions with its physical type)
709        let s = Schema::parse_str(r#"["string", {"type":"string","logicalType":"uuid"}]"#).unwrap();
710        expect![[r#"
711            Union(
712                UnionSchema {
713                    schemas: [
714                        String,
715                        Uuid,
716                    ],
717                    variant_index: {
718                        String: 0,
719                        Uuid: 1,
720                    },
721                },
722            )
723        "#]]
724        .assert_debug_eq(&s);
725        // Note: Java Avro lib rejects this (logical type unions with its physical type)
726        let s = Schema::parse_str(r#"["int", {"type":"int", "logicalType": "date"}]"#).unwrap();
727        expect![[r#"
728            Union(
729                UnionSchema {
730                    schemas: [
731                        Int,
732                        Date,
733                    ],
734                    variant_index: {
735                        Int: 0,
736                        Date: 1,
737                    },
738                },
739            )
740        "#]]
741        .assert_debug_eq(&s);
742        // Note: Java Avro lib allows this (2 decimal with different "name")
743        let s = Schema::parse_str(
744            r#"[
745{"type":"fixed","name":"Decimal128","size":16,"logicalType":"decimal","precision":38,"scale":2},
746{"type":"fixed","name":"Decimal256","size":32,"logicalType":"decimal","precision":50,"scale":2}
747]"#,
748        );
749        expect![[r#"
750            Err(
751                Error {
752                    details: Unions cannot contain duplicate types,
753                },
754            )
755        "#]]
756        .assert_debug_eq(&s);
757    }
758
759    #[test]
760    fn test_avro_lib_union_record_bug() {
761        // multiple named types (record)
762        let s = Schema::parse_str(
763            r#"
764    {
765      "type": "record",
766      "name": "Root",
767      "fields": [
768        {
769          "name": "unionTypeComplex",
770          "type": [
771            "null",
772            {"type": "record", "name": "Email","fields": [{"name":"inner","type":"string"}]},
773            {"type": "record", "name": "Fax","fields": [{"name":"inner","type":"int"}]},
774            {"type": "record", "name": "Sms","fields": [{"name":"inner","type":"int"}]}
775          ]
776        }
777      ]
778    }
779        "#,
780        )
781        .unwrap();
782
783        let bytes = hex::decode("060c").unwrap();
784        // Correct should be variant 3 (Sms)
785        let correct_value = from_avro_datum(&s, &mut bytes.as_slice(), None);
786        expect![[r#"
787                Ok(
788                    Record(
789                        [
790                            (
791                                "unionTypeComplex",
792                                Union(
793                                    3,
794                                    Record(
795                                        [
796                                            (
797                                                "inner",
798                                                Int(
799                                                    6,
800                                                ),
801                                            ),
802                                        ],
803                                    ),
804                                ),
805                            ),
806                        ],
807                    ),
808                )
809            "#]]
810        .assert_debug_eq(&correct_value);
811        // Bug: We got variant 2 (Fax) here, if we pass the reader schema.
812        let wrong_value = from_avro_datum(&s, &mut bytes.as_slice(), Some(&s));
813        expect![[r#"
814                Ok(
815                    Record(
816                        [
817                            (
818                                "unionTypeComplex",
819                                Union(
820                                    2,
821                                    Record(
822                                        [
823                                            (
824                                                "inner",
825                                                Int(
826                                                    6,
827                                                ),
828                                            ),
829                                        ],
830                                    ),
831                                ),
832                            ),
833                        ],
834                    ),
835                )
836            "#]]
837        .assert_debug_eq(&wrong_value);
838
839        // The bug below can explain what happened.
840        // The two records below are actually incompatible: https://avro.apache.org/docs/1.11.1/specification/_print/#schema-resolution
841        // > both schemas are records with the _same (unqualified) name_
842        // In from_avro_datum, it first reads the value with the writer schema, and then
843        // it just uses the reader schema to interpret the value.
844        // The value doesn't have record "name" information. So it wrongly passed the conversion.
845        // The correct way is that we need to use both the writer and reader schema in the second step to interpret the value.
846
847        let s = Schema::parse_str(
848            r#"
849    {
850      "type": "record",
851      "name": "Root",
852      "fields": [
853        {
854          "name": "a",
855          "type": "int"
856        }
857      ]
858    }
859        "#,
860        )
861        .unwrap();
862        let s2 = Schema::parse_str(
863            r#"
864{
865  "type": "record",
866  "name": "Root222",
867  "fields": [
868    {
869      "name": "a",
870      "type": "int"
871    }
872  ]
873}
874    "#,
875        )
876        .unwrap();
877
878        let bytes = hex::decode("0c").unwrap();
879        let value = from_avro_datum(&s, &mut bytes.as_slice(), Some(&s2));
880        expect![[r#"
881            Ok(
882                Record(
883                    [
884                        (
885                            "a",
886                            Int(
887                                6,
888                            ),
889                        ),
890                    ],
891                ),
892            )
893        "#]]
894        .assert_debug_eq(&value);
895    }
896
897    #[test]
898    fn test_convert_decimal() {
899        // 280
900        let v = vec![1, 24];
901        let avro_decimal = AvroDecimal::from(v);
902        let rust_decimal = scaled_bigint_to_rust_decimal(avro_decimal.into(), 0).unwrap();
903        assert_eq!(rust_decimal, rust_decimal::Decimal::from(280));
904
905        // 28.1
906        let v = vec![1, 25];
907        let avro_decimal = AvroDecimal::from(v);
908        let rust_decimal = scaled_bigint_to_rust_decimal(avro_decimal.into(), 1).unwrap();
909        assert_eq!(rust_decimal, rust_decimal::Decimal::try_from(28.1).unwrap());
910
911        // 1.1234567891
912        let value = BigInt::from(11234567891_i64);
913        let decimal = scaled_bigint_to_rust_decimal(value, 10).unwrap();
914        assert_eq!(
915            decimal,
916            rust_decimal::Decimal::try_from(1.1234567891).unwrap()
917        );
918
919        // 1.123456789123456789123456789
920        let v = vec![3, 161, 77, 58, 146, 180, 49, 220, 100, 4, 95, 21];
921        let avro_decimal = AvroDecimal::from(v);
922        let rust_decimal = scaled_bigint_to_rust_decimal(avro_decimal.into(), 27).unwrap();
923        assert_eq!(
924            rust_decimal,
925            rust_decimal::Decimal::from_str("1.123456789123456789123456789").unwrap()
926        );
927    }
928
929    /// Convert Avro value to datum.For now, support the following [Avro type](https://avro.apache.org/docs/current/spec.html).
930    ///  - boolean
931    ///  - int : i32
932    ///  - long: i64
933    ///  - float: f32
934    ///  - double: f64
935    ///  - string: String
936    ///  - Date (the number of days from the unix epoch, 1970-1-1 UTC)
937    ///  - Timestamp (the number of milliseconds from the unix epoch,  1970-1-1 00:00:00.000 UTC)
938    fn from_avro_value(
939        value: Value,
940        value_schema: &Schema,
941        shape: &DataType,
942    ) -> anyhow::Result<Datum> {
943        Ok(AvroParseOptions::create(value_schema)
944            .inner
945            .convert_to_datum(value_schema, &value, shape)?
946            .to_owned_datum())
947    }
948
949    #[test]
950    fn test_avro_timestamptz_micros() {
951        let v1 = Value::TimestampMicros(1620000000000000);
952        let v2 = Value::TimestampMillis(1620000000000);
953        let value_schema1 = Schema::TimestampMicros;
954        let value_schema2 = Schema::TimestampMillis;
955        let datum1 = from_avro_value(v1, &value_schema1, &DataType::Timestamptz).unwrap();
956        let datum2 = from_avro_value(v2, &value_schema2, &DataType::Timestamptz).unwrap();
957        assert_eq!(
958            datum1,
959            Some(ScalarImpl::Timestamptz(
960                Timestamptz::from_str("2021-05-03T00:00:00Z").unwrap()
961            ))
962        );
963        assert_eq!(
964            datum2,
965            Some(ScalarImpl::Timestamptz(
966                Timestamptz::from_str("2021-05-03T00:00:00Z").unwrap()
967            ))
968        );
969    }
970
971    #[test]
972    fn test_decimal_truncate() {
973        let schema = Schema::parse_str(
974            r#"
975            {
976                "type": "bytes",
977                "logicalType": "decimal",
978                "precision": 38,
979                "scale": 18
980            }
981            "#,
982        )
983        .unwrap();
984        let bytes = vec![0x3f, 0x3f, 0x3f, 0x3f, 0x3f, 0x3f, 0x3f];
985        let value = Value::Decimal(AvroDecimal::from(bytes));
986        let resp = from_avro_value(value, &schema, &DataType::Decimal).unwrap();
987        assert_eq!(
988            resp,
989            Some(ScalarImpl::Decimal(Decimal::Normalized(
990                rust_decimal::Decimal::from_str("0.017802464409370431").unwrap()
991            )))
992        );
993    }
994
995    #[test]
996    fn test_variable_scale_decimal() {
997        let schema = Schema::parse_str(
998            r#"
999            {
1000                "type": "record",
1001                "name": "VariableScaleDecimal",
1002                "namespace": "io.debezium.data",
1003                "fields": [
1004                    {
1005                        "name": "scale",
1006                        "type": "int"
1007                    },
1008                    {
1009                        "name": "value",
1010                        "type": "bytes"
1011                    }
1012                ]
1013            }
1014            "#,
1015        )
1016        .unwrap();
1017        let value = Value::Record(vec![
1018            ("scale".to_owned(), Value::Int(0)),
1019            ("value".to_owned(), Value::Bytes(vec![0x01, 0x02, 0x03])),
1020        ]);
1021
1022        let resp = from_avro_value(value, &schema, &DataType::Decimal).unwrap();
1023        assert_eq!(resp, Some(ScalarImpl::Decimal(Decimal::from(66051))));
1024    }
1025}
risingwave_connector_codec/decoder/avro/mod.rs

risingwave_connector_codec/decoder/avro/
mod.rs
