Repository: risingwavelabs/risingwave
File: src/connector/src/parser/unified/json.rs (Rust)

Code before:
                        .decode(value_str)
                        .map_err(|_| create_error())?
                        .into_boxed_slice()
                        .into(),
                }
            }
            // Handle Debezium PostGIS geometry type: {"srid": <int>, "wkb": <base64_string>}
            // We extract the wkb field and decode it as EWKB bytes
            (DataType::Bytea, ValueType::Object) => {
                let obj = value.as_object().ok_or_else(create_error)?;

                // Check if this is a Debezium geometry object by verifying both fields exist
                let srid_value = obj.get("srid");
                let wkb_value = obj.get("wkb");

                if let (Some(srid), Some(wkb)) = (srid_value, wkb_value) {
                    // Validate srid is numeric and wkb is string
                    if srid.as_i64().is_some() && wkb.as_str().is_some() {
                        // Extract and decode the wkb field (base64-encoded EWKB)
                        let wkb_str = wkb.as_str().unwrap();
                        base64::engine::general_purpose::STANDARD
                            .decode(wkb_str)
                            .map_err(|_| create_error())?
                            .into_boxed_slice()
                            .into()
                    } else {
                        Err(create_error())?
                    }
                } else {
                    // Not a geometry object, reject
                    Err(create_error())?
                }
            }
            // ---- Jsonb -----
            (DataType::Jsonb, ValueType::String)
                if matches!(self.json_value_handling, JsonValueHandling::AsString) =>
            {
                // Check if this value is the Debezium unavailable value (TOAST handling for postgres-cdc).
                // Debezium will base64 encode the bytea type placeholder.
                // When a placeholder is encountered, it is converted into a jsonb format placeholder to match the original type.
                match self.handle_toast_columns {
                    true => JsonbVal::from_debezium_unavailable_value(value.as_str().unwrap())
                        .map_err(|_| create_error())?
                        .into(),
                    false => JsonbVal::from_str(value.as_str().unwrap())
                        .map_err(|_| create_error())?
                        .into(),
                }
            }
            (DataType::Jsonb, _)
                if matches!(self.json_value_handling, JsonValueHandling::AsValue) =>

Review comment (refactor):
Extract the actual parsing logic into a dedicated function, for 2 reasons:
* Easier extension for MySQL in the future as we may need to combine srid and wkb to get ewkb. (This [SMT](https://debezium.io/documentation/reference/stable/transformations/geometry-format-transformer.html) may also be useful in the future for unification).
* Keep the caller `match`/`if` focused on dispatching. There may be other `object` that maps to `bytea` but not a `geometry(numeric srid+string wkb)`

Code after:
                    );
                }
                StructValue::new(fields).into()
            }

            // String containing json object, e.g. "{\"a\": 1, \"b\": 2}"
            // Try to parse it as json object.
            (DataType::Struct(_), ValueType::String)
                if matches!(self.struct_handling, StructHandling::AllowJsonString) =>
            {
                // TODO: avoid copy by accepting `&mut BorrowedValue` in `parse` method.
                let mut value = value.as_str().unwrap().as_bytes().to_vec();
                let value =
                    simd_json::to_borrowed_value(&mut value[..]).map_err(|_| create_error())?;
                return self
                    .parse(&value, type_expected)
                    .map(|d| d.to_owned_datum().into());
            }

            // ---- List -----
            (DataType::List(list_type), ValueType::Array) => ListValue::new({
                let item_type = list_type.elem();
                let array = value.as_array().unwrap();
                let mut builder = item_type.create_array_builder(array.len());
                for v in array {
                    let value = self.parse(v, item_type)?;
                    builder.append(value);
                }
                builder.finish()
            })
            .into(),

            // ---- Bytea -----
            (DataType::Bytea, ValueType::String) => {
                let value_str = value.as_str().unwrap();

                match self.bytea_handling {
                    ByteaHandling::Standard => {
                        let mut buf = Vec::new();
                        str_to_bytea(value_str, &mut buf).map_err(|_| create_error())?;
                        buf.into()
                    }
                    ByteaHandling::Base64 => base64::engine::general_purpose::STANDARD
                        .decode(value_str)
                        .map_err(|_| create_error())?
                        .into_boxed_slice()
                        .into(),
                }
            }
            // Handle Debezium PostGIS geometry type: {"srid": <int>, "wkb": <base64_string>}
            // We extract the wkb field and decode it as EWKB bytes