fix debezium

Signed-off-by: xxchan <[email protected]>

src/connector/codec/src/decoder/avro/mod.rs

@@ -337,7 +337,8 @@ impl Access for AvroAccess<'_> {
         let mut options: AvroParseOptions<'_> = self.options.clone();
 
         debug_assert!(
-            path.len() == 1 || (path.len() == 2 && path[0] == "before"),
+            path.len() == 1
+                || (path.len() == 2 && matches!(path[0], "before" | "after" | "source")),
             "unexpected path access: {:?}",
             path
         );

src/connector/codec/src/decoder/mod.rs

@@ -67,7 +67,9 @@ pub trait Access {
     ///
     /// What `path` to access is decided by the CDC layer, i.e., the `FORMAT ...` part (`ChangeEvent`).
     /// e.g.,
-    /// - `DebeziumChangeEvent` accesses `["before", "col_name"]` for value, `["op"]` for op type.
+    /// - `DebeziumChangeEvent` accesses `["before", "col_name"]` for value,
+    ///   `["source", "db"]`, `["source", "table"]` etc. for additional columns' values,
+    ///   `["op"]` for op type.
     /// - `MaxwellChangeEvent` accesses `["data", "col_name"]` for value, `["type"]` for op type.
     /// - In the simplest case, for `FORMAT PLAIN/UPSERT` (`KvEvent`), they just access `["col_name"]` for value, and op type is derived.
     ///

src/connector/src/parser/debezium/avro_parser.rs

@@ -149,6 +149,14 @@ impl DebeziumAvroParserConfig {
         //     "default": null
         // },
         // ...]
+
+        // Other fields are:
+        // - source: describes the source metadata for the event
+        // - op
+        // - ts_ms
+        // - transaction
+        // See <https://debezium.io/documentation/reference/stable/connectors/mysql.html#mysql-events>
+
         avro_schema_to_column_descs(
             avro_schema_skip_nullable_union(avro_extract_field_schema(
                 // FIXME: use resolved schema here.