github-actions[bot] commented on code in PR #4362:
URL: https://github.com/apache/iggy/pull/4362#discussion_r4161213879
##########
core/connectors/sdk/src/convert.rs:
##########
@@ -49,11 +51,38 @@ pub fn owned_value_to_serde_json(value:
&simd_json::OwnedValue) -> serde_json::V
}
}
+/// Renames each key of a `Payload::Json` object via `rename`; other payloads
pass through. Collisions keep the later entry while the map is Vec-backed,
unspecified once hash-backed.
+pub(crate) fn apply_field_mappings(
+ payload: Payload,
+ mut rename: impl FnMut(String) -> String,
+) -> Payload {
+ match payload {
+ Payload::Json(json_value) => {
+ if let simd_json::OwnedValue::Object(mut map) = json_value {
+ let renamed: Vec<(String, simd_json::OwnedValue)> = map
+ .drain()
+ .map(|(key, value)| (rename(key), value))
+ .collect();
+ for (key, value) in renamed {
Review Comment:
warning: When two keys rename to one name, `apply_field_mappings` silently
drops one of the two values. Log a warning on the collision, or reject the
mapping when the config loads.
##########
core/connectors/sdk/src/encoders/avro.rs:
##########
@@ -444,6 +425,20 @@ mod tests {
assert!(result.is_ok());
let encoded_data = result.unwrap();
assert!(!encoded_data.is_empty());
+
+ // Decode the real bytes and check the value; schema strictness alone
isn't proof the rename worked.
+ let schema = AvroSchema::parse_str(&schema_json).unwrap();
+ let mut reader: &[u8] = &encoded_data;
+ let decoded = GenericDatumReader::builder(&schema)
+ .build()
+ .unwrap()
+ .read_value(&mut reader)
+ .unwrap();
+ let decoded_json = serde_json::Value::try_from(decoded).unwrap();
+
+ assert_eq!(decoded_json["name"], serde_json::json!("Alice"));
+ assert_eq!(decoded_json["age"], serde_json::json!(30));
+ assert!(decoded_json.get("full_name").is_none());
Review Comment:
simplification: `decoded_json` holds only the schema fields `name` and
`age`, so this assertion can never fail. Delete the line; the equality check on
line 439 already proves the rename.
##########
core/connectors/sdk/src/encoders/proto.rs:
##########
@@ -279,37 +279,17 @@ impl ProtoStreamEncoder {
Ok(())
}
- fn apply_field_transformations(&self, payload: Payload) -> Result<Payload,
Error> {
+ fn apply_field_transformations(&self, payload: Payload) -> Payload {
if let Some(mappings) = &self.config.field_mappings {
- match payload {
- Payload::Json(json_value) => {
- if let simd_json::OwnedValue::Object(mut map) = json_value
{
- let mut new_entries = Vec::new();
-
- for (key, value) in map.iter() {
- let proto_key = mappings
- .iter()
- .find(|(_, json_name)| *json_name == key)
- .map(|(proto_name, _)| proto_name.clone())
- .unwrap_or_else(|| key.clone());
-
- new_entries.push((proto_key, value.clone()));
- }
-
- map.clear();
- for (key, value) in new_entries {
- map.insert(key, value);
- }
-
- Ok(Payload::Json(simd_json::OwnedValue::Object(map)))
- } else {
- Ok(Payload::Json(json_value))
- }
- }
- other => Ok(other),
- }
+ apply_field_mappings(payload, |key| {
+ mappings
+ .iter()
+ .find(|(_, json_name)| json_name.as_str() == key.as_str())
Review Comment:
warning: The proto encoder renames to an arbitrary proto field when two
mappings share a JSON name, so a restart can change the wire field. Sort the
candidates, or reject duplicate JSON names at config load.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
