This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new c5f21c0c2 perf(connectors): drain field_mappings instead of
clone-then-clear (#4362)
c5f21c0c2 is described below
commit c5f21c0c2d6931b957cbdf7211ed3f5cc82326c9
Author: Matthew Patton <[email protected]>
AuthorDate: Sat Oct 10 16:09:08 2026 -0400
perf(connectors): drain field_mappings instead of clone-then-clear (#4362)
---
core/connectors/sdk/src/convert.rs | 287 ++++++++++++++++++++-
core/connectors/sdk/src/decoders/avro.rs | 35 +--
core/connectors/sdk/src/decoders/flatbuffer.rs | 35 +--
core/connectors/sdk/src/decoders/proto.rs | 64 +----
core/connectors/sdk/src/encoders/avro.rs | 54 ++--
core/connectors/sdk/src/encoders/flatbuffer.rs | 48 ++--
core/connectors/sdk/src/encoders/proto.rs | 91 ++++---
core/connectors/sdk/src/transforms/avro_convert.rs | 36 +--
.../connectors/sdk/src/transforms/proto_convert.rs | 31 +--
.../connectors/sdk/tests/flatbuffer_integration.rs | 16 ++
10 files changed, 435 insertions(+), 262 deletions(-)
diff --git a/core/connectors/sdk/src/convert.rs
b/core/connectors/sdk/src/convert.rs
index 6fe4dd1bd..993e052b4 100644
--- a/core/connectors/sdk/src/convert.rs
+++ b/core/connectors/sdk/src/convert.rs
@@ -15,10 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-//! Value conversion utilities for connector sinks.
+//! Value conversion utilities for connectors.
//!
-//! Provides shared conversion functions between serialization formats used by
-//! the connector ecosystem (e.g., `simd_json` ↔ `serde_json`).
+//! Shared conversion and payload-transform functions used by the connector
+//! ecosystem: `simd_json`/`serde_json` bridging, JSON key renaming.
+
+use crate::Payload;
/// Convert `simd_json::OwnedValue` to `serde_json::Value` via direct
structural mapping.
///
@@ -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 {
+ map.insert(key, value);
+ }
+ Payload::Json(simd_json::OwnedValue::Object(map))
+ } else {
+ Payload::Json(json_value)
+ }
+ }
+ other => other,
+ }
+}
+
#[cfg(test)]
mod tests {
- use super::*;
+ use std::collections::HashMap;
+
use simd_json::{OwnedValue, StaticNode};
+ use super::*;
+
#[test]
fn test_null() {
assert_eq!(
@@ -161,4 +190,254 @@ mod tests {
serde_json::json!({"a": {"b": "v"}})
);
}
+
+ #[test]
+ fn apply_field_mappings_should_rename_mapped_key_and_keep_unmapped_key() {
+ let mappings = HashMap::from([("old_field".to_string(),
"new_field".to_string())]);
+
+ let payload = Payload::Json(simd_json::json!({
+ "old_field": "renamed",
+ "unchanged_field": "stays_same"
+ }));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.get("new_field").unwrap(),
&OwnedValue::from("renamed"));
+ assert!(!map.contains_key("old_field"));
+ assert_eq!(
+ map.get("unchanged_field").unwrap(),
+ &OwnedValue::from("stays_same")
+ );
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn
apply_field_mappings_should_swap_two_fields_without_losing_either_value() {
+ let mappings = HashMap::from([
+ ("a".to_string(), "b".to_string()),
+ ("b".to_string(), "a".to_string()),
+ ]);
+
+ let payload = Payload::Json(simd_json::json!({
+ "a": 1,
+ "b": 2
+ }));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.get("a").unwrap(), &OwnedValue::from(2));
+ assert_eq!(map.get("b").unwrap(), &OwnedValue::from(1));
+ assert_eq!(map.len(), 2);
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn
apply_field_mappings_should_rename_chained_fields_without_a_pre_existing_target()
{
+ let mappings = HashMap::from([
+ ("a".to_string(), "b".to_string()),
+ ("b".to_string(), "c".to_string()),
+ ]);
+ let payload = Payload::Json(simd_json::json!({"a": 1, "b": 2}));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.get("b").unwrap(), &OwnedValue::from(1));
+ assert_eq!(map.get("c").unwrap(), &OwnedValue::from(2));
+ assert_eq!(map.len(), 2);
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn apply_field_mappings_should_only_rename_top_level_keys() {
+ let mappings = HashMap::from([("a".to_string(),
"renamed".to_string())]);
+ let payload = Payload::Json(simd_json::json!({
+ "a": {"a": "nested_value"}
+ }));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert!(!map.contains_key("a"));
+ if let OwnedValue::Object(nested) = map.get("renamed").unwrap() {
+ assert_eq!(nested.get("a").unwrap(),
&OwnedValue::from("nested_value"));
+ } else {
+ panic!("Expected nested JSON object");
+ }
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn apply_field_mappings_should_allow_self_mapping_as_a_no_op() {
+ let mappings = HashMap::from([("a".to_string(), "a".to_string())]);
+ let payload = Payload::Json(simd_json::json!({"a": 1}));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.get("a").unwrap(), &OwnedValue::from(1));
+ assert_eq!(map.len(), 1);
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn apply_field_mappings_should_keep_last_value_when_rename_collides() {
+ let mappings = HashMap::from([("a".to_string(), "c".to_string())]);
+ let payload = Payload::Json(simd_json::json!({"a": 1, "c": 2}));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.len(), 1);
+ assert_eq!(map.get("c").unwrap(), &OwnedValue::from(2));
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn
apply_field_mappings_should_collapse_a_pre_existing_duplicate_key_to_its_last_value()
{
+ // Raw bytes, not `json!` (which dedupes at macro-build time), so the
real parser's duplicate-key handling is exercised.
+ let mut bytes = br#"{"a":1,"a":2,"b":3}"#.to_vec();
+ let value: OwnedValue = simd_json::to_owned_value(&mut bytes).unwrap();
+ if let OwnedValue::Object(map) = &value {
+ assert_eq!(
+ map.len(),
+ 3,
+ "expected both raw \"a\" entries to survive parsing"
+ );
+ } else {
+ panic!("Expected JSON object");
+ }
+ let payload = Payload::Json(value);
+
+ let result = apply_field_mappings(payload, |key| key);
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.len(), 2);
+ assert_eq!(map.get("a").unwrap(), &OwnedValue::from(2));
+ assert_eq!(map.get("b").unwrap(), &OwnedValue::from(3));
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn
apply_field_mappings_should_rename_and_preserve_every_value_above_the_32_key_vec_threshold()
+ {
+ // 40 keys forces simd_json's backing map past the 32-key Vec
threshold.
+ let mut json = String::from("{");
+ for i in 0..39 {
+ json.push_str(&format!(r#""field_{i}":{i},"#));
+ }
+ json.push_str(r#""field_39":39}"#);
+ let mut bytes = json.into_bytes();
+ let value: OwnedValue = simd_json::to_owned_value(&mut bytes).unwrap();
+ if let OwnedValue::Object(map) = &value {
+ assert!(map.is_map(), "expected the hashbrown backend above 32
keys");
+ } else {
+ panic!("Expected JSON object");
+ }
+ let payload = Payload::Json(value);
+
+ let mappings: HashMap<String, String> = (0..5)
+ .map(|i| (format!("field_{i}"), format!("renamed_{i}")))
+ .collect();
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.len(), 40);
+ for i in 0..40 {
+ let key = if i < 5 {
+ format!("renamed_{i}")
+ } else {
+ format!("field_{i}")
+ };
+ assert_eq!(
+ map.get(key.as_str())
+ .unwrap_or_else(|| panic!("missing {key}")),
+ &OwnedValue::from(i as i64),
+ "wrong value for {key}"
+ );
+ }
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn apply_field_mappings_should_ignore_rename_when_old_key_absent() {
+ let mappings = HashMap::from([("missing".to_string(),
"new_name".to_string())]);
+ let payload = Payload::Json(simd_json::json!({"a": 1}));
+
+ let result =
+ apply_field_mappings(payload, |key|
mappings.get(&key).cloned().unwrap_or(key));
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert!(!map.contains_key("new_name"));
+ assert_eq!(map.get("a").unwrap(), &OwnedValue::from(1));
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn
apply_field_mappings_should_leave_populated_object_unchanged_when_renames_empty()
{
+ let payload = Payload::Json(simd_json::json!({"a": 1, "b": 2}));
+
+ let result = apply_field_mappings(payload, |key| key);
+
+ if let Payload::Json(OwnedValue::Object(map)) = result {
+ assert_eq!(map.get("a").unwrap(), &OwnedValue::from(1));
+ assert_eq!(map.get("b").unwrap(), &OwnedValue::from(2));
+ assert_eq!(map.len(), 2);
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
+ #[test]
+ fn apply_field_mappings_should_leave_non_object_json_unchanged() {
+ let payload = Payload::Json(simd_json::json!([1, 2, 3]));
+
+ let result = apply_field_mappings(payload, |key| key);
+
+ if let Payload::Json(json_value) = result {
+ assert_eq!(json_value, simd_json::json!([1, 2, 3]));
+ } else {
+ panic!("Expected JSON payload");
+ }
+ }
+
+ #[test]
+ fn apply_field_mappings_should_leave_non_json_payload_unchanged() {
+ let payload = Payload::Raw(vec![1, 2, 3]);
+
+ let result = apply_field_mappings(payload, |key| key);
+
+ if let Payload::Raw(bytes) = result {
+ assert_eq!(bytes, vec![1, 2, 3]);
+ } else {
+ panic!("Expected Raw payload");
+ }
+ }
}
diff --git a/core/connectors/sdk/src/decoders/avro.rs
b/core/connectors/sdk/src/decoders/avro.rs
index 7a0e519c8..039c2b609 100644
--- a/core/connectors/sdk/src/decoders/avro.rs
+++ b/core/connectors/sdk/src/decoders/avro.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{Error, Payload, Schema, StreamDecoder};
+use crate::{Error, Payload, Schema, StreamDecoder,
convert::apply_field_mappings};
use apache_avro::Schema as AvroSchema;
use apache_avro::reader::datum::GenericDatumReader;
use serde::{Deserialize, Serialize};
@@ -166,43 +166,18 @@ impl AvroStreamDecoder {
Error::CannotDecode(Schema::Avro)
})?;
- let transformed =
self.apply_field_transformations(Payload::Json(json_value))?;
- Ok(transformed)
+ Ok(self.apply_field_transformations(Payload::Json(json_value)))
}
fn decode_as_raw(&self, payload: Vec<u8>) -> Result<Payload, Error> {
Ok(Payload::Avro(payload))
}
- 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() {
- if let Some(new_key) = mappings.get(key) {
- new_entries.push((new_key.clone(),
value.clone()));
- } else {
- new_entries.push((key.clone(), 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.get(&key).cloned().unwrap_or(key))
} else {
- Ok(payload)
+ payload
}
}
}
diff --git a/core/connectors/sdk/src/decoders/flatbuffer.rs
b/core/connectors/sdk/src/decoders/flatbuffer.rs
index 54f70509c..ec2f13ff4 100644
--- a/core/connectors/sdk/src/decoders/flatbuffer.rs
+++ b/core/connectors/sdk/src/decoders/flatbuffer.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{Error, Payload, Schema, StreamDecoder};
+use crate::{Error, Payload, Schema, StreamDecoder,
convert::apply_field_mappings};
use base64::Engine;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
@@ -106,8 +106,7 @@ impl FlatBufferStreamDecoder {
);
let json_payload = simd_json::OwnedValue::Object(json_map);
- let transformed =
self.apply_field_transformations(Payload::Json(json_payload))?;
- Ok(transformed)
+ Ok(self.apply_field_transformations(Payload::Json(json_payload)))
}
fn decode_as_raw(&self, data: Vec<u8>) -> Result<Payload, Error> {
@@ -117,35 +116,11 @@ impl FlatBufferStreamDecoder {
Ok(Payload::FlatBuffer(data))
}
- 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() {
- if let Some(new_key) = mappings.get(key) {
- new_entries.push((new_key.clone(),
value.clone()));
- } else {
- new_entries.push((key.clone(), 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.get(&key).cloned().unwrap_or(key))
} else {
- Ok(payload)
+ payload
}
}
}
diff --git a/core/connectors/sdk/src/decoders/proto.rs
b/core/connectors/sdk/src/decoders/proto.rs
index a4d774a6e..faed2ac8c 100644
--- a/core/connectors/sdk/src/decoders/proto.rs
+++ b/core/connectors/sdk/src/decoders/proto.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{Error, Payload, Schema, StreamDecoder};
+use crate::{Error, Payload, Schema, StreamDecoder,
convert::apply_field_mappings};
use base64::Engine;
use prost::Message;
use prost_types::Any;
@@ -521,35 +521,11 @@ impl ProtoStreamDecoder {
.ok_or(Error::InvalidProtobufPayload)
}
- 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() {
- if let Some(new_key) = mappings.get(key) {
- new_entries.push((new_key.clone(),
value.clone()));
- } else {
- new_entries.push((key.clone(), 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.get(&key).cloned().unwrap_or(key))
} else {
- Ok(payload)
+ payload
}
}
@@ -586,7 +562,7 @@ impl StreamDecoder for ProtoStreamDecoder {
let decoded_payload = self.decode_with_schema(payload)?;
- self.apply_field_transformations(decoded_payload)
+ Ok(self.apply_field_transformations(decoded_payload))
}
}
@@ -655,40 +631,28 @@ mod tests {
#[test]
fn decode_should_apply_field_mappings_when_configured() {
+ // use_any_wrapper alone always emits literal type_url/value keys,
making a mapping unobservable, so decode via a real schema instead.
let mut field_mappings = HashMap::new();
- field_mappings.insert("old_field".to_string(),
"new_field".to_string());
- field_mappings.insert("user_id".to_string(), "id".to_string());
+ field_mappings.insert("id".to_string(), "user_id".to_string());
let config = ProtoConfig {
+ schema_path: Some(PathBuf::from("examples/user.proto")),
+ message_type: Some("com.example.User".to_string()),
+ use_any_wrapper: false,
field_mappings: Some(field_mappings),
- use_any_wrapper: true,
..ProtoConfig::default()
};
let decoder = ProtoStreamDecoder::new(config);
- let json_content = simd_json::json!({
- "old_field": "should_be_renamed",
- "user_id": 123,
- "unchanged_field": "stays_same"
- });
- let json_string = simd_json::to_string(&json_content).unwrap();
-
- let any = Any {
- type_url: "type.googleapis.com/custom.Message".to_string(),
- value: json_string.into_bytes(),
- };
+ // Hand-encode a second, unmapped field (name) alongside id so the
rename isn't the message's only field.
+ let mut encoded = 42i32.encode_to_vec();
+ encoded.extend_from_slice(&[0x12, 3, b'B', b'o', b'b']);
- let encoded = any.encode_to_vec();
let result = decoder.decode(encoded);
assert!(result.is_ok());
if let Ok(Payload::Json(json_value)) = result {
- if let simd_json::OwnedValue::Object(map) = &json_value {
- assert!(map.contains_key("type_url"));
- assert!(map.contains_key("value"));
- } else {
- panic!("Expected JSON object");
- }
+ assert_eq!(json_value, simd_json::json!({"user_id": 42, "name":
"Bob"}));
} else {
panic!("Expected JSON payload");
}
diff --git a/core/connectors/sdk/src/encoders/avro.rs
b/core/connectors/sdk/src/encoders/avro.rs
index 1d6701390..588207fc9 100644
--- a/core/connectors/sdk/src/encoders/avro.rs
+++ b/core/connectors/sdk/src/encoders/avro.rs
@@ -15,7 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{Error, Payload, Schema, StreamEncoder,
convert::owned_value_to_serde_json};
+use crate::{
+ Error, Payload, Schema, StreamEncoder,
+ convert::{apply_field_mappings, owned_value_to_serde_json},
+};
use apache_avro::Schema as AvroSchema;
use apache_avro::writer::datum::GenericDatumWriter;
use base64::Engine;
@@ -118,35 +121,11 @@ impl AvroStreamEncoder {
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() {
- if let Some(new_key) = mappings.get(key) {
- new_entries.push((new_key.clone(),
value.clone()));
- } else {
- new_entries.push((key.clone(), 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.get(&key).cloned().unwrap_or(key))
} else {
- Ok(payload)
+ payload
}
}
@@ -352,7 +331,7 @@ impl StreamEncoder for AvroStreamEncoder {
}
fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error> {
- let transformed_payload = self.apply_field_transformations(payload)?;
+ let transformed_payload = self.apply_field_transformations(payload);
match transformed_payload {
Payload::Json(json_value) => self.encode_json_to_avro(json_value),
@@ -373,6 +352,8 @@ impl Default for AvroStreamEncoder {
#[cfg(test)]
mod tests {
+ use apache_avro::reader::datum::GenericDatumReader;
+
use super::*;
fn create_test_schema_json() -> String {
@@ -428,7 +409,7 @@ mod tests {
field_mappings.insert("full_name".to_string(), "name".to_string());
let config = AvroEncoderConfig {
- schema_json: Some(schema_json),
+ schema_json: Some(schema_json.clone()),
field_mappings: Some(field_mappings),
..AvroEncoderConfig::default()
};
@@ -444,6 +425,19 @@ 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));
}
#[test]
diff --git a/core/connectors/sdk/src/encoders/flatbuffer.rs
b/core/connectors/sdk/src/encoders/flatbuffer.rs
index 32c60daaf..07187c358 100644
--- a/core/connectors/sdk/src/encoders/flatbuffer.rs
+++ b/core/connectors/sdk/src/encoders/flatbuffer.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{Error, Payload, Schema, StreamEncoder};
+use crate::{Error, Payload, Schema, StreamEncoder,
convert::apply_field_mappings};
use base64::Engine;
use flatbuffers::FlatBufferBuilder;
use serde::{Deserialize, Serialize};
@@ -63,35 +63,11 @@ impl FlatBufferStreamEncoder {
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() {
- if let Some(new_key) = mappings.get(key) {
- new_entries.push((new_key.clone(),
value.clone()));
- } else {
- new_entries.push((key.clone(), 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.get(&key).cloned().unwrap_or(key))
} else {
- Ok(payload)
+ payload
}
}
@@ -229,7 +205,7 @@ impl StreamEncoder for FlatBufferStreamEncoder {
}
fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error> {
- let transformed_payload = self.apply_field_transformations(payload)?;
+ let transformed_payload = self.apply_field_transformations(payload);
match transformed_payload {
Payload::Json(json_value) =>
self.encode_json_to_flatbuffer(json_value),
@@ -327,6 +303,20 @@ mod tests {
assert!(result.is_ok());
let encoded_data = result.unwrap();
assert!(!encoded_data.is_empty());
+
+ // No `.fbs` schema to decode typed fields, but keys are plain UTF-8
strings inline, so the rename is checkable byte-wise.
+ assert!(contains_bytes(&encoded_data, b"new_field"));
+ assert!(!contains_bytes(&encoded_data, b"old_field"));
+ assert!(contains_bytes(&encoded_data, b"unchanged_field"));
+ // Confirms the renamed key kept its value rather than being dropped.
+ assert!(contains_bytes(&encoded_data, b"should_be_renamed"));
+ assert!(contains_bytes(&encoded_data, b"stays_same"));
+ }
+
+ fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
+ haystack
+ .windows(needle.len())
+ .any(|window| window == needle)
}
#[test]
diff --git a/core/connectors/sdk/src/encoders/proto.rs
b/core/connectors/sdk/src/encoders/proto.rs
index 44dca94f0..acac89b82 100644
--- a/core/connectors/sdk/src/encoders/proto.rs
+++ b/core/connectors/sdk/src/encoders/proto.rs
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-use crate::{Error, Payload, Schema, StreamEncoder};
+use crate::{Error, Payload, Schema, StreamEncoder,
convert::apply_field_mappings};
use base64::{Engine as Base64Engine, engine::general_purpose};
use iggy_common::IggyTimestamp;
use prost::Message;
@@ -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())
+ .map(|(proto_name, _)| proto_name.clone())
+ .unwrap_or(key)
+ })
} else {
- Ok(payload)
+ payload
}
}
@@ -730,7 +710,7 @@ impl StreamEncoder for ProtoStreamEncoder {
}
fn encode(&self, payload: Payload) -> Result<Vec<u8>, Error> {
- let transformed_payload = self.apply_field_transformations(payload)?;
+ let transformed_payload = self.apply_field_transformations(payload);
self.encode_with_schema(transformed_payload)
}
@@ -909,9 +889,18 @@ mod tests {
let decoded_json = simd_json::to_owned_value(&mut
encoded_bytes_mut).unwrap();
if let simd_json::OwnedValue::Object(map) = decoded_json {
- assert!(map.contains_key("user_id"));
- assert!(map.contains_key("full_name"));
- assert!(map.contains_key("email"));
+ assert_eq!(
+ map.get("user_id").unwrap(),
+ &simd_json::OwnedValue::from(123)
+ );
+ assert_eq!(
+ map.get("full_name").unwrap(),
+ &simd_json::OwnedValue::from("John Doe")
+ );
+ assert_eq!(
+ map.get("email").unwrap(),
+ &simd_json::OwnedValue::from("[email protected]")
+ );
assert!(!map.contains_key("id"));
assert!(!map.contains_key("name"));
} else {
@@ -919,6 +908,40 @@ mod tests {
}
}
+ #[test]
+ fn
encode_should_resolve_ambiguous_reverse_lookup_to_one_of_the_colliding_keys() {
+ // Reverse-lookup collision resolves via HashMap iteration order (not
deterministic), so only pin that one of the two survives.
+ let mut field_mappings = HashMap::new();
+ field_mappings.insert("proto_a".to_string(), "shared".to_string());
+ field_mappings.insert("proto_b".to_string(), "shared".to_string());
+
+ let config = ProtoEncoderConfig {
+ field_mappings: Some(field_mappings),
+ use_any_wrapper: false,
+ ..ProtoEncoderConfig::default()
+ };
+ let encoder = ProtoStreamEncoder::new_with_config(config);
+
+ let json_value = simd_json::json!({"shared": "value"});
+ let result = encoder.encode(Payload::Json(json_value));
+
+ assert!(result.is_ok());
+ let mut encoded_bytes = result.unwrap();
+ let decoded_json = simd_json::to_owned_value(&mut
encoded_bytes).unwrap();
+
+ if let simd_json::OwnedValue::Object(map) = decoded_json {
+ assert_eq!(map.len(), 1);
+ let renamed_to_a = map.get("proto_a") ==
Some(&simd_json::OwnedValue::from("value"));
+ let renamed_to_b = map.get("proto_b") ==
Some(&simd_json::OwnedValue::from("value"));
+ assert!(
+ renamed_to_a || renamed_to_b,
+ "expected the value to land on proto_a or proto_b, got {map:?}"
+ );
+ } else {
+ panic!("Expected JSON object");
+ }
+ }
+
#[test]
fn encode_should_use_custom_type_url_prefix_when_configured() {
let config = ProtoEncoderConfig {
diff --git a/core/connectors/sdk/src/transforms/avro_convert.rs
b/core/connectors/sdk/src/transforms/avro_convert.rs
index 5a14dd000..02062715c 100644
--- a/core/connectors/sdk/src/transforms/avro_convert.rs
+++ b/core/connectors/sdk/src/transforms/avro_convert.rs
@@ -22,7 +22,7 @@ use std::path::PathBuf;
use super::{Transform, TransformType};
use crate::decoders::avro::{AvroConfig, AvroStreamDecoder};
use crate::encoders::avro::{AvroEncoderConfig, AvroStreamEncoder};
-use crate::{DecodedMessage, Error, Payload, Schema, TopicMetadata};
+use crate::{DecodedMessage, Error, Payload, Schema, TopicMetadata,
convert::apply_field_mappings};
use crate::{StreamDecoder, StreamEncoder};
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -68,36 +68,14 @@ impl AvroConvert {
Self::new(AvroConvertConfig::default())
}
- fn apply_field_mappings(
+ fn apply_field_transformations(
&self,
payload: Payload,
field_mappings: &HashMap<String, String>,
- ) -> Result<Payload, Error> {
- 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() {
- if let Some(new_key) = field_mappings.get(key) {
- new_entries.push((new_key.clone(), value.clone()));
- } else {
- new_entries.push((key.clone(), 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),
- }
+ ) -> Payload {
+ apply_field_mappings(payload, |key| {
+ field_mappings.get(&key).cloned().unwrap_or(key)
+ })
}
}
@@ -112,7 +90,7 @@ impl Transform for AvroConvert {
mut message: DecodedMessage,
) -> Result<Option<DecodedMessage>, Error> {
if let Some(field_mappings) = &self.config.field_mappings {
- message.payload = self.apply_field_mappings(message.payload,
field_mappings)?;
+ message.payload =
self.apply_field_transformations(message.payload, field_mappings);
}
message.payload = match (&self.config.source_format,
&self.config.target_format) {
diff --git a/core/connectors/sdk/src/transforms/proto_convert.rs
b/core/connectors/sdk/src/transforms/proto_convert.rs
index 4f239641f..da32a9b40 100644
--- a/core/connectors/sdk/src/transforms/proto_convert.rs
+++ b/core/connectors/sdk/src/transforms/proto_convert.rs
@@ -26,7 +26,7 @@ use tracing::{error, info};
use super::{Transform, TransformType};
use crate::encoders::proto::ProtoStreamEncoder;
-use crate::{DecodedMessage, Error, Payload, Schema, TopicMetadata};
+use crate::{DecodedMessage, Error, Payload, Schema, TopicMetadata,
convert::apply_field_mappings};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ProtoConvertConfig {
@@ -484,32 +484,11 @@ impl ProtoConvert {
}
}
- 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 new_key =
mappings.get(key).cloned().unwrap_or_else(|| key.clone());
- new_entries.push((new_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.get(&key).cloned().unwrap_or(key))
} else {
- Ok(payload)
+ payload
}
}
@@ -732,7 +711,7 @@ impl Transform for ProtoConvert {
_metadata: &TopicMetadata,
mut message: DecodedMessage,
) -> Result<Option<DecodedMessage>, Error> {
- let transformed_payload =
self.apply_field_transformations(message.payload)?;
+ let transformed_payload =
self.apply_field_transformations(message.payload);
let converted_payload = match (self.config.source_format,
self.config.target_format) {
(Schema::Proto, _) =>
self.convert_from_protobuf(transformed_payload)?,
diff --git a/core/connectors/sdk/tests/flatbuffer_integration.rs
b/core/connectors/sdk/tests/flatbuffer_integration.rs
index ab14669e1..b360744b5 100644
--- a/core/connectors/sdk/tests/flatbuffer_integration.rs
+++ b/core/connectors/sdk/tests/flatbuffer_integration.rs
@@ -78,12 +78,24 @@ async fn
should_transform_with_real_schema_and_field_mapping() {
bytes.len() >= 4,
"FlatBuffer should have minimum header size"
);
+
+ // "id"/"name" are substrings of "user_id"/"full_name", so
absence of the old keys is the reliable signal, not presence of the new ones.
+ assert!(!contains_bytes(&bytes, b"user_id"));
+ assert!(!contains_bytes(&bytes, b"full_name"));
+ assert!(contains_bytes(&bytes, b"email"));
+ assert!(contains_bytes(&bytes, b"Jane Smith"));
}
other => panic!("Expected FlatBuffer payload, got: {other:?}"),
}
}
}
+fn contains_bytes(haystack: &[u8], needle: &[u8]) -> bool {
+ haystack
+ .windows(needle.len())
+ .any(|window| window == needle)
+}
+
#[tokio::test]
async fn should_encode_decode_with_schema_configuration() {
let encoder_config = FlatBufferEncoderConfig {
@@ -225,6 +237,10 @@ async fn
should_transform_json_to_flatbuffer_with_field_mappings() {
Payload::FlatBuffer(bytes) => {
println!("Transformed FlatBuffer: {} bytes", bytes.len());
assert!(!bytes.is_empty(), "FlatBuffer should not be empty");
+ assert!(!contains_bytes(&bytes, b"user_id"));
+ assert!(!contains_bytes(&bytes, b"full_name"));
+ assert!(contains_bytes(&bytes, b"email"));
+ assert!(contains_bytes(&bytes, b"Jane Smith"));
}
other => panic!("Expected FlatBuffer payload, got: {other:?}"),
}
