From 18817101737eec861641914b0befa7e34822eed6 Mon Sep 17 00:00:00 2001 From: StandingMan Date: Wed, 22 Jul 2026 22:56:07 +0800 Subject: [PATCH 1/2] feat(connectors): add FlatBuffer BFBS schema loader Signed-off-by: StandingMan --- Cargo.lock | 13 + Cargo.toml | 1 + core/connectors/sdk/Cargo.toml | 1 + core/connectors/sdk/src/flatbuffer/mod.rs | 23 + core/connectors/sdk/src/flatbuffer/schema.rs | 1381 +++++++++++++++++ core/connectors/sdk/src/lib.rs | 1 + .../sdk/tests/fixtures/flatbuffer/types.bfbs | Bin 0 -> 2312 bytes .../sdk/tests/fixtures/flatbuffer/types.fbs | 67 + .../sdk/tests/fixtures/flatbuffer/user.bfbs | Bin 0 -> 1100 bytes licenserc.toml | 1 + 10 files changed, 1488 insertions(+) create mode 100644 core/connectors/sdk/src/flatbuffer/mod.rs create mode 100644 core/connectors/sdk/src/flatbuffer/schema.rs create mode 100644 core/connectors/sdk/tests/fixtures/flatbuffer/types.bfbs create mode 100644 core/connectors/sdk/tests/fixtures/flatbuffer/types.fbs create mode 100644 core/connectors/sdk/tests/fixtures/flatbuffer/user.bfbs diff --git a/Cargo.lock b/Cargo.lock index ea71b9e89d..944d8c42ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4998,6 +4998,18 @@ dependencies = [ "rustc_version", ] +[[package]] +name = "flatbuffers-reflection" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c29b83e11dbe84a256acfb8f3ea3637204a6b1b08257cbf1b7dfd48f8ef0904" +dependencies = [ + "anyhow", + "flatbuffers", + "num-traits", + "thiserror 1.0.69", +] + [[package]] name = "flate2" version = "1.1.9" @@ -7165,6 +7177,7 @@ dependencies = [ "base64", "dashmap", "flatbuffers", + "flatbuffers-reflection", "http 1.4.2", "humantime", "iggy", diff --git a/Cargo.toml b/Cargo.toml index 8d363b0050..2743b02868 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -169,6 +169,7 @@ figlet-rs = "1.0.0" figment = { version = "0.10.19", features = ["toml", "env"] } file-operation = "0.8.28" flatbuffers = "25.12.19" +flatbuffers-reflection = "0.1.0" flume = "0.12.0" fs2 = "0.4.3" futures = "0.3.33" diff --git a/core/connectors/sdk/Cargo.toml b/core/connectors/sdk/Cargo.toml index fcd2b6f2ae..f680752e1a 100644 --- a/core/connectors/sdk/Cargo.toml +++ b/core/connectors/sdk/Cargo.toml @@ -43,6 +43,7 @@ async-trait = { workspace = true } base64 = { workspace = true } dashmap = { workspace = true } flatbuffers = { workspace = true } +flatbuffers-reflection = { workspace = true } http = { workspace = true } humantime = { workspace = true } iggy = { workspace = true } diff --git a/core/connectors/sdk/src/flatbuffer/mod.rs b/core/connectors/sdk/src/flatbuffer/mod.rs new file mode 100644 index 0000000000..bbecce9bdd --- /dev/null +++ b/core/connectors/sdk/src/flatbuffer/mod.rs @@ -0,0 +1,23 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +mod schema; + +pub use schema::{ + EnumDescriptor, EnumValueDescriptor, FieldDescriptor, FlatBufferSchema, ObjectDescriptor, + SchemaFeatures, TypeDescriptor, TypeKind, +}; diff --git a/core/connectors/sdk/src/flatbuffer/schema.rs b/core/connectors/sdk/src/flatbuffer/schema.rs new file mode 100644 index 0000000000..4a4d258841 --- /dev/null +++ b/core/connectors/sdk/src/flatbuffer/schema.rs @@ -0,0 +1,1381 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::{BTreeMap, BTreeSet}; +use std::fs; +use std::path::Path; +use std::sync::Arc; + +use flatbuffers::{FILE_IDENTIFIER_LENGTH, SIZE_UOFFSET}; +use flatbuffers_reflection::reflection::{ + AdvancedFeatures, BaseType, Field, Schema, Type, root_as_schema, schema_buffer_has_identifier, +}; + +use crate::Error; + +const MAX_BFBS_SIZE_BYTES: usize = 16 * 1024 * 1024; +const MAX_OBJECTS: usize = 65_536; +const MAX_FIELDS_PER_OBJECT: usize = 65_536; +const MAX_ENUMS: usize = 65_536; +const MAX_VALUES_PER_ENUM: usize = 65_536; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SchemaFeatures { + bits: u64, +} + +impl SchemaFeatures { + pub fn bits(self) -> u64 { + self.bits + } + + pub fn advanced_arrays(self) -> bool { + self.bits & AdvancedFeatures::AdvancedArrayFeatures.bits() != 0 + } + + pub fn advanced_unions(self) -> bool { + self.bits & AdvancedFeatures::AdvancedUnionFeatures.bits() != 0 + } + + pub fn optional_scalars(self) -> bool { + self.bits & AdvancedFeatures::OptionalScalars.bits() != 0 + } + + pub fn default_vectors_and_strings(self) -> bool { + self.bits & AdvancedFeatures::DefaultVectorsAndStrings.bits() != 0 + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TypeKind { + None, + UnionType, + Bool, + Int8, + Uint8, + Int16, + Uint16, + Int32, + Uint32, + Int64, + Uint64, + Float32, + Float64, + String, + Vector, + Object, + Union, + Array, + Vector64, + Unknown(i8), +} + +impl From for TypeKind { + fn from(base_type: BaseType) -> Self { + match base_type { + BaseType::None => Self::None, + BaseType::UType => Self::UnionType, + BaseType::Bool => Self::Bool, + BaseType::Byte => Self::Int8, + BaseType::UByte => Self::Uint8, + BaseType::Short => Self::Int16, + BaseType::UShort => Self::Uint16, + BaseType::Int => Self::Int32, + BaseType::UInt => Self::Uint32, + BaseType::Long => Self::Int64, + BaseType::ULong => Self::Uint64, + BaseType::Float => Self::Float32, + BaseType::Double => Self::Float64, + BaseType::String => Self::String, + BaseType::Vector => Self::Vector, + BaseType::Obj => Self::Object, + BaseType::Union => Self::Union, + BaseType::Array => Self::Array, + BaseType::Vector64 => Self::Vector64, + unknown => Self::Unknown(unknown.0), + } + } +} + +impl TypeKind { + pub fn is_scalar(self) -> bool { + matches!( + self, + Self::UnionType + | Self::Bool + | Self::Int8 + | Self::Uint8 + | Self::Int16 + | Self::Uint16 + | Self::Int32 + | Self::Uint32 + | Self::Int64 + | Self::Uint64 + | Self::Float32 + | Self::Float64 + ) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TypeDescriptor { + base_type: TypeKind, + element_type: TypeKind, + index: Option, + fixed_length: u16, + base_size: u32, + element_size: u32, +} + +impl TypeDescriptor { + pub fn base_type(&self) -> TypeKind { + self.base_type + } + + pub fn element_type(&self) -> TypeKind { + self.element_type + } + + pub fn index(&self) -> Option { + self.index + } + + pub fn fixed_length(&self) -> u16 { + self.fixed_length + } + + pub fn base_size(&self) -> u32 { + self.base_size + } + + pub fn element_size(&self) -> u32 { + self.element_size + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct FieldDescriptor { + name: String, + id: u16, + offset: u16, + field_type: TypeDescriptor, + default_integer: i64, + default_real: f64, + deprecated: bool, + required: bool, + key: bool, + optional: bool, + padding: u16, + offset64: bool, +} + +impl FieldDescriptor { + pub fn name(&self) -> &str { + &self.name + } + + pub fn id(&self) -> u16 { + self.id + } + + pub fn offset(&self) -> u16 { + self.offset + } + + pub fn field_type(&self) -> &TypeDescriptor { + &self.field_type + } + + pub fn default_integer(&self) -> i64 { + self.default_integer + } + + pub fn default_real(&self) -> f64 { + self.default_real + } + + pub fn is_deprecated(&self) -> bool { + self.deprecated + } + + pub fn is_required(&self) -> bool { + self.required + } + + pub fn is_key(&self) -> bool { + self.key + } + + pub fn is_optional(&self) -> bool { + self.optional + } + + pub fn padding(&self) -> u16 { + self.padding + } + + pub fn uses_64_bit_offset(&self) -> bool { + self.offset64 + } +} + +#[derive(Debug, Clone, PartialEq)] +pub struct ObjectDescriptor { + /// Fully qualified table or struct name from the BFBS schema. + name: String, + /// Position of the object in the BFBS `objects` vector. + index: usize, + /// Whether the object uses the fixed-layout FlatBuffer struct representation. + is_struct: bool, + /// Owned metadata for fields declared by the object. + fields: Vec, + /// Field name to `fields` index lookup. + field_indices: BTreeMap, +} + +impl ObjectDescriptor { + pub fn name(&self) -> &str { + &self.name + } + + pub fn index(&self) -> usize { + self.index + } + + pub fn is_struct(&self) -> bool { + self.is_struct + } + + pub fn fields(&self) -> &[FieldDescriptor] { + &self.fields + } + + pub fn field(&self, name: &str) -> Option<&FieldDescriptor> { + self.field_indices + .get(name) + .and_then(|index| self.fields.get(*index)) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EnumValueDescriptor { + name: String, + value: i64, + union_type: Option, +} + +impl EnumValueDescriptor { + pub fn name(&self) -> &str { + &self.name + } + + pub fn value(&self) -> i64 { + self.value + } + + pub fn union_type(&self) -> Option<&TypeDescriptor> { + self.union_type.as_ref() + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct EnumDescriptor { + name: String, + index: usize, + is_union: bool, + underlying_type: TypeDescriptor, + values: Vec, + value_indices: BTreeMap, + number_indices: BTreeMap, +} + +impl EnumDescriptor { + pub fn name(&self) -> &str { + &self.name + } + + pub fn index(&self) -> usize { + self.index + } + + pub fn is_union(&self) -> bool { + self.is_union + } + + pub fn underlying_type(&self) -> &TypeDescriptor { + &self.underlying_type + } + + pub fn values(&self) -> &[EnumValueDescriptor] { + &self.values + } + + pub fn value(&self, name: &str) -> Option<&EnumValueDescriptor> { + self.value_indices + .get(name) + .and_then(|index| self.values.get(*index)) + } + + pub fn value_by_number(&self, value: i64) -> Option<&EnumValueDescriptor> { + self.number_indices + .get(&value) + .and_then(|index| self.values.get(*index)) + } +} + +#[derive(Debug, Clone)] +pub struct FlatBufferSchema { + /// Verified BFBS bytes retained for creating short-lived reflection views. + bytes: Arc<[u8]>, + /// Index of the root table selected from the schema or explicit configuration. + root_table_index: usize, + /// Optional four-byte identifier expected in encoded business payloads. + file_identifier: Option<[u8; 4]>, + /// Owned metadata for every table and struct declared by the schema. + objects: Vec, + /// Fully qualified object name to `objects` index lookup. + object_indices: BTreeMap, + /// Owned metadata for every enum and union declared by the schema. + enums: Vec, + /// Fully qualified enum or union name to `enums` index lookup. + enum_indices: BTreeMap, + /// Advanced schema features recognized by this loader version. + features: SchemaFeatures, +} + +impl FlatBufferSchema { + pub fn from_path( + schema_path: impl AsRef, + root_table_name: Option<&str>, + ) -> Result { + let schema_path = schema_path.as_ref(); + let schema_size = fs::metadata(schema_path) + .map_err(|error| { + Error::InvalidConfigValue(format!( + "Failed to inspect FlatBuffer BFBS schema '{}': {error}", + schema_path.display() + )) + })? + .len(); + if schema_size > MAX_BFBS_SIZE_BYTES as u64 { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema '{}' exceeds the {MAX_BFBS_SIZE_BYTES}-byte size limit", + schema_path.display() + ))); + } + let bytes = fs::read(schema_path).map_err(|error| { + Error::InvalidConfigValue(format!( + "Failed to read FlatBuffer BFBS schema '{}': {error}", + schema_path.display() + )) + })?; + + Self::from_bytes(bytes, root_table_name) + } + + pub fn from_bytes( + bytes: impl Into>, + root_table_name: Option<&str>, + ) -> Result { + let bytes = bytes.into(); + let schema = parse_schema(&bytes)?; + let features = parse_schema_features(&schema)?; + let (objects, object_indices) = collect_objects(&schema)?; + let (enums, enum_indices) = collect_enums(&schema)?; + validate_references(&objects, &enums)?; + let root_table_index = + resolve_root_table(&schema, &objects, &object_indices, root_table_name)?; + let file_identifier = parse_file_identifier(&schema)?; + + Ok(Self { + bytes, + root_table_index, + file_identifier, + objects, + object_indices, + enums, + enum_indices, + features, + }) + } + + pub fn bytes(&self) -> &[u8] { + &self.bytes + } + + pub fn reflection_schema(&self) -> Result, Error> { + parse_schema(&self.bytes) + } + + pub fn root_table(&self) -> &ObjectDescriptor { + &self.objects[self.root_table_index] + } + + pub fn file_identifier(&self) -> Option<[u8; 4]> { + self.file_identifier + } + + pub fn objects(&self) -> &[ObjectDescriptor] { + &self.objects + } + + pub fn object(&self, name: &str) -> Option<&ObjectDescriptor> { + self.object_indices + .get(name) + .and_then(|index| self.objects.get(*index)) + } + + pub fn object_by_index(&self, index: usize) -> Option<&ObjectDescriptor> { + self.objects.get(index) + } + + pub fn enums(&self) -> &[EnumDescriptor] { + &self.enums + } + + pub fn enum_descriptor(&self, name: &str) -> Option<&EnumDescriptor> { + self.enum_indices + .get(name) + .and_then(|index| self.enums.get(*index)) + } + + pub fn enum_by_index(&self, index: usize) -> Option<&EnumDescriptor> { + self.enums.get(index) + } + + pub fn features(&self) -> SchemaFeatures { + self.features + } +} + +fn parse_schema(bytes: &[u8]) -> Result, Error> { + if bytes.len() > MAX_BFBS_SIZE_BYTES { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema exceeds the {MAX_BFBS_SIZE_BYTES}-byte size limit" + ))); + } + if bytes.len() < SIZE_UOFFSET + FILE_IDENTIFIER_LENGTH || !schema_buffer_has_identifier(bytes) { + return Err(Error::InvalidConfigValue( + "FlatBuffer schema is missing the BFBS file identifier".to_string(), + )); + } + + root_as_schema(bytes).map_err(|error| { + Error::InvalidConfigValue(format!("Invalid FlatBuffer BFBS schema: {error}")) + }) +} + +fn parse_schema_features(schema: &Schema<'_>) -> Result { + let features = schema.advanced_features(); + let known_bits = AdvancedFeatures::AdvancedArrayFeatures.bits() + | AdvancedFeatures::AdvancedUnionFeatures.bits() + | AdvancedFeatures::OptionalScalars.bits() + | AdvancedFeatures::DefaultVectorsAndStrings.bits(); + let unknown_bits = features.bits() & !known_bits; + if unknown_bits != 0 { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema uses unknown advanced feature bits 0x{unknown_bits:x}" + ))); + } + + Ok(SchemaFeatures { + bits: features.bits(), + }) +} + +fn collect_objects( + schema: &Schema<'_>, +) -> Result<(Vec, BTreeMap), Error> { + let schema_objects = schema.objects(); + if schema_objects.len() > MAX_OBJECTS { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema contains more than {MAX_OBJECTS} objects" + ))); + } + let mut objects = Vec::with_capacity(schema_objects.len()); + let mut object_indices = BTreeMap::new(); + + for index in 0..schema_objects.len() { + let object = schema_objects.get(index); + let name = object.name().to_string(); + if name.is_empty() { + return Err(Error::InvalidConfigValue( + "FlatBuffer BFBS schema contains an object with an empty name".to_string(), + )); + } + if object_indices.insert(name.clone(), index).is_some() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema contains duplicate object '{name}'" + ))); + } + let (fields, field_indices) = collect_fields(&name, object.fields())?; + objects.push(ObjectDescriptor { + name, + index, + is_struct: object.is_struct(), + fields, + field_indices, + }); + } + + Ok((objects, object_indices)) +} + +fn collect_fields( + object_name: &str, + schema_fields: flatbuffers::Vector<'_, flatbuffers::ForwardsUOffset>>, +) -> Result<(Vec, BTreeMap), Error> { + if schema_fields.len() > MAX_FIELDS_PER_OBJECT { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer object '{object_name}' contains more than {MAX_FIELDS_PER_OBJECT} fields" + ))); + } + let mut fields = Vec::with_capacity(schema_fields.len()); + let mut field_indices = BTreeMap::new(); + let mut field_ids = BTreeSet::new(); + let mut field_offsets = BTreeSet::new(); + let mut has_key = false; + + for index in 0..schema_fields.len() { + let field = schema_fields.get(index); + let name = field.name().to_string(); + if name.is_empty() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer object '{object_name}' contains a field with an empty name" + ))); + } + if field_indices.insert(name.clone(), index).is_some() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer object '{object_name}' contains duplicate field '{name}'" + ))); + } + if !field_ids.insert(field.id()) { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer object '{object_name}' contains duplicate field ID {}", + field.id() + ))); + } + if !field_offsets.insert(field.offset()) { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer object '{object_name}' contains duplicate field offset {}", + field.offset() + ))); + } + if field.required() && field.optional() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer field '{object_name}.{name}' cannot be both required and optional" + ))); + } + let type_context = format!("field '{object_name}.{name}'"); + let field_type = describe_type(field.type_(), &type_context)?; + if field.key() { + if has_key { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer object '{object_name}' defines more than one key field" + ))); + } + if !field_type.base_type().is_scalar() && field_type.base_type() != TypeKind::String { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {type_context} uses key on an unsupported type" + ))); + } + has_key = true; + } + fields.push(FieldDescriptor { + name, + id: field.id(), + offset: field.offset(), + field_type, + default_integer: field.default_integer(), + default_real: field.default_real(), + deprecated: field.deprecated(), + required: field.required(), + key: field.key(), + optional: field.optional(), + padding: field.padding(), + offset64: field.offset64(), + }); + } + + Ok((fields, field_indices)) +} + +fn collect_enums( + schema: &Schema<'_>, +) -> Result<(Vec, BTreeMap), Error> { + let schema_enums = schema.enums(); + if schema_enums.len() > MAX_ENUMS { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema contains more than {MAX_ENUMS} enums or unions" + ))); + } + let mut enums = Vec::with_capacity(schema_enums.len()); + let mut enum_indices = BTreeMap::new(); + + for index in 0..schema_enums.len() { + let schema_enum = schema_enums.get(index); + let name = schema_enum.name().to_string(); + if name.is_empty() { + return Err(Error::InvalidConfigValue( + "FlatBuffer BFBS schema contains an enum or union with an empty name".to_string(), + )); + } + if enum_indices.insert(name.clone(), index).is_some() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer BFBS schema contains duplicate enum or union '{name}'" + ))); + } + + let schema_values = schema_enum.values(); + if schema_values.len() > MAX_VALUES_PER_ENUM { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer enum or union '{name}' contains more than {MAX_VALUES_PER_ENUM} values" + ))); + } + let mut values = Vec::with_capacity(schema_values.len()); + let mut value_indices = BTreeMap::new(); + let mut number_indices = BTreeMap::new(); + for value_index in 0..schema_values.len() { + let schema_value = schema_values.get(value_index); + let value_name = schema_value.name().to_string(); + if value_name.is_empty() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer enum or union '{name}' contains a value with an empty name" + ))); + } + if value_indices + .insert(value_name.clone(), value_index) + .is_some() + { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer enum or union '{name}' contains duplicate value '{value_name}'" + ))); + } + let union_type = schema_value + .union_type() + .map(|field_type| { + describe_type( + field_type, + &format!("enum or union value '{name}.{}'", schema_value.name()), + ) + }) + .transpose()? + .filter(|field_type| field_type.base_type() != TypeKind::None); + values.push(EnumValueDescriptor { + name: value_name, + value: schema_value.value(), + union_type, + }); + number_indices + .entry(schema_value.value()) + .or_insert(value_index); + } + + let underlying_type = describe_type( + schema_enum.underlying_type(), + &format!("enum or union '{name}'"), + )?; + enums.push(EnumDescriptor { + name, + index, + is_union: schema_enum.is_union(), + underlying_type, + values, + value_indices, + number_indices, + }); + } + + Ok((enums, enum_indices)) +} + +fn describe_type(field_type: Type<'_>, context: &str) -> Result { + let index = match field_type.index() { + -1 => None, + index if index >= 0 => Some(index as usize), + index => { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} contains invalid type index {index}" + ))); + } + }; + let descriptor = TypeDescriptor { + base_type: field_type.base_type().into(), + element_type: field_type.element().into(), + index, + fixed_length: field_type.fixed_length(), + base_size: field_type.base_size(), + element_size: field_type.element_size(), + }; + if let TypeKind::Unknown(value) = descriptor.base_type { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} uses unknown base type {value}" + ))); + } + if let TypeKind::Unknown(value) = descriptor.element_type { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} uses unknown element type {value}" + ))); + } + + Ok(descriptor) +} + +fn validate_references( + objects: &[ObjectDescriptor], + enums: &[EnumDescriptor], +) -> Result<(), Error> { + for object in objects { + for field in object.fields() { + validate_type_reference( + field.field_type(), + objects, + enums, + &format!("field '{}.{}'", object.name(), field.name()), + )?; + } + } + + for enum_descriptor in enums { + validate_enum_underlying_type(enum_descriptor)?; + for value in enum_descriptor.values() { + if let Some(union_type) = value.union_type() { + validate_type_reference( + union_type, + objects, + enums, + &format!( + "enum or union value '{}.{}'", + enum_descriptor.name(), + value.name() + ), + )?; + } + } + } + + Ok(()) +} + +fn validate_enum_underlying_type(enum_descriptor: &EnumDescriptor) -> Result<(), Error> { + let base_type = enum_descriptor.underlying_type().base_type(); + let valid = if enum_descriptor.is_union() { + base_type == TypeKind::UnionType + } else { + matches!( + base_type, + TypeKind::Int8 + | TypeKind::Uint8 + | TypeKind::Int16 + | TypeKind::Uint16 + | TypeKind::Int32 + | TypeKind::Uint32 + | TypeKind::Int64 + | TypeKind::Uint64 + ) + }; + if !valid { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer enum or union '{}' has invalid underlying type {base_type:?}", + enum_descriptor.name() + ))); + } + + Ok(()) +} + +fn validate_type_reference( + field_type: &TypeDescriptor, + objects: &[ObjectDescriptor], + enums: &[EnumDescriptor], + context: &str, +) -> Result<(), Error> { + if field_type.base_type() == TypeKind::None { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} does not define a base type" + ))); + } + if field_type.base_type() == TypeKind::Array && field_type.fixed_length() == 0 { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} defines a fixed array with zero length" + ))); + } + if field_type.base_type() != TypeKind::Array && field_type.fixed_length() != 0 { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} defines a fixed length for a non-array type" + ))); + } + + match field_type.base_type() { + TypeKind::Object => validate_object_index(field_type.index(), objects, context), + TypeKind::Union | TypeKind::UnionType => { + validate_enum_index(field_type.index(), enums, true, context) + } + TypeKind::Vector | TypeKind::Vector64 | TypeKind::Array => { + match field_type.element_type() { + TypeKind::None => Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} does not define a container element type" + ))), + TypeKind::Object => validate_object_index(field_type.index(), objects, context), + TypeKind::Union | TypeKind::UnionType => { + validate_enum_index(field_type.index(), enums, true, context) + } + _ => match field_type.index() { + Some(_) => validate_enum_index(field_type.index(), enums, false, context), + None => Ok(()), + }, + } + } + _ => match field_type.index() { + Some(_) => validate_enum_index(field_type.index(), enums, false, context), + None => Ok(()), + }, + } +} + +fn validate_object_index( + index: Option, + objects: &[ObjectDescriptor], + context: &str, +) -> Result<(), Error> { + let index = index.ok_or_else(|| { + Error::InvalidConfigValue(format!( + "FlatBuffer {context} is missing its object type index" + )) + })?; + if objects.get(index).is_none() { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} references missing object index {index}" + ))); + } + + Ok(()) +} + +fn validate_enum_index( + index: Option, + enums: &[EnumDescriptor], + require_union: bool, + context: &str, +) -> Result<(), Error> { + let index = index.ok_or_else(|| { + Error::InvalidConfigValue(format!( + "FlatBuffer {context} is missing its enum or union type index" + )) + })?; + let enum_descriptor = enums.get(index).ok_or_else(|| { + Error::InvalidConfigValue(format!( + "FlatBuffer {context} references missing enum or union index {index}" + )) + })?; + if enum_descriptor.is_union() != require_union { + let expected = if require_union { "union" } else { "enum" }; + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer {context} references '{}' as an {expected}", + enum_descriptor.name() + ))); + } + + Ok(()) +} + +fn resolve_root_table( + schema: &Schema<'_>, + objects: &[ObjectDescriptor], + object_indices: &BTreeMap, + configured_name: Option<&str>, +) -> Result { + let root_name = match configured_name { + Some(name) if !name.trim().is_empty() => name, + Some(_) => { + return Err(Error::InvalidConfigValue( + "FlatBuffer root table name cannot be empty".to_string(), + )); + } + None => schema + .root_table() + .map(|object| object.name()) + .ok_or_else(|| { + Error::InvalidConfigValue( + "FlatBuffer BFBS schema does not define a root table".to_string(), + ) + })?, + }; + + let index = object_indices.get(root_name).copied().ok_or_else(|| { + Error::InvalidConfigValue(format!( + "FlatBuffer root table '{root_name}' was not found in the BFBS schema" + )) + })?; + let root_table = &objects[index]; + if root_table.is_struct { + return Err(Error::InvalidConfigValue(format!( + "FlatBuffer root object '{}' is a struct, not a table", + root_table.name + ))); + } + + Ok(index) +} + +fn parse_file_identifier(schema: &Schema<'_>) -> Result, Error> { + let Some(identifier) = schema.file_ident() else { + return Ok(None); + }; + if identifier.is_empty() { + return Ok(None); + } + let identifier = identifier.as_bytes().try_into().map_err(|_| { + Error::InvalidConfigValue(format!( + "FlatBuffer file identifier must contain exactly 4 bytes, got {}", + identifier.len() + )) + })?; + + Ok(Some(identifier)) +} + +#[cfg(test)] +mod tests { + use flatbuffers::{FlatBufferBuilder, WIPOffset}; + use flatbuffers_reflection::reflection::{ + AdvancedFeatures, Enum, Field, FieldArgs, Object, ObjectArgs, Schema, SchemaArgs, Type, + TypeArgs, finish_schema_buffer, + }; + + use super::*; + + #[test] + fn given_valid_bfbs_when_loaded_should_cache_schema_metadata() { + let bytes = build_bfbs( + &[("example.User", false), ("example.Address", false)], + 0, + Some("USER"), + ); + + let schema = FlatBufferSchema::from_bytes(bytes, None).expect("schema should load"); + + assert_eq!(schema.root_table().name(), "example.User"); + assert_eq!(schema.root_table().index(), 0); + assert_eq!(schema.file_identifier(), Some(*b"USER")); + assert_eq!(schema.objects().len(), 2); + assert_eq!( + schema + .object("example.Address") + .map(|object| object.index()), + Some(1) + ); + assert!(schema.reflection_schema().is_ok()); + } + + #[test] + fn given_real_bfbs_file_when_loaded_should_read_schema_metadata() { + // Precompiled from examples/user.fbs so CI verifies real flatc output without requiring + // the compiler to be installed. + let schema_path = + Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/flatbuffer/user.bfbs"); + + let schema = FlatBufferSchema::from_path(schema_path, None) + .expect("real BFBS schema should load from path"); + let reflection_schema = schema + .reflection_schema() + .expect("cached BFBS schema should remain valid"); + let root_table = reflection_schema + .root_table() + .expect("real BFBS schema should define a root table"); + + assert_eq!(schema.root_table().name(), "com.example.User"); + assert_eq!(root_table.fields().len(), 7); + assert!(schema.object("com.example.Address").is_some()); + assert!(schema.object("com.example.UserList").is_some()); + assert_eq!(schema.file_identifier(), None); + } + + #[test] + fn given_real_bfbs_with_supported_types_when_loaded_should_cache_type_metadata() { + let schema_path = + Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/flatbuffer/types.bfbs"); + + let schema = FlatBufferSchema::from_path(schema_path, None) + .expect("real BFBS type schema should load from path"); + let root = schema.root_table(); + + assert_eq!(root.name(), "test.types.AllTypes"); + assert_eq!(schema.file_identifier(), Some(*b"TYPE")); + assert!(schema.features().optional_scalars()); + assert!(!schema.features().advanced_arrays()); + assert_field_type(root, "bool_value", TypeKind::Bool); + assert_field_type(root, "int8_value", TypeKind::Int8); + assert_field_type(root, "uint8_value", TypeKind::Uint8); + assert_field_type(root, "int16_value", TypeKind::Int16); + assert_field_type(root, "uint16_value", TypeKind::Uint16); + assert_field_type(root, "int32_value", TypeKind::Int32); + assert_field_type(root, "uint32_value", TypeKind::Uint32); + assert_field_type(root, "int64_value", TypeKind::Int64); + assert_field_type(root, "uint64_value", TypeKind::Uint64); + assert_field_type(root, "float32_value", TypeKind::Float32); + assert_field_type(root, "float64_value", TypeKind::Float64); + assert_field_type(root, "string_value", TypeKind::String); + + let bool_field = root.field("bool_value").expect("bool field should exist"); + assert_eq!(bool_field.default_integer(), 1); + let string_field = root + .field("string_value") + .expect("string field should exist"); + assert!(string_field.is_required()); + let optional_field = root + .field("optional_score") + .expect("optional scalar field should exist"); + assert!(optional_field.is_optional()); + + let bytes_field = root.field("bytes").expect("byte vector should exist"); + assert_eq!(bytes_field.field_type().base_type(), TypeKind::Vector); + assert_eq!(bytes_field.field_type().element_type(), TypeKind::Uint8); + + let point_field = root.field("point").expect("struct field should exist"); + assert_eq!(point_field.field_type().base_type(), TypeKind::Object); + let point = schema + .object_by_index( + point_field + .field_type() + .index() + .expect("object field should have a schema index"), + ) + .expect("object index should resolve"); + assert_eq!(point.name(), "test.types.Point"); + assert!(point.is_struct()); + + let status_field = root.field("statuses").expect("enum vector should exist"); + let status = schema + .enum_by_index( + status_field + .field_type() + .index() + .expect("enum vector should have a schema index"), + ) + .expect("enum index should resolve"); + assert_eq!(status.name(), "test.types.Status"); + assert!(!status.is_union()); + assert_eq!(status.underlying_type().base_type(), TypeKind::Uint8); + assert_eq!( + status.value("Active").map(EnumValueDescriptor::value), + Some(1) + ); + assert_eq!( + status.value_by_number(2).map(EnumValueDescriptor::name), + Some("Disabled") + ); + + let contact = schema + .enum_descriptor("test.types.Contact") + .expect("union descriptor should exist"); + assert!(contact.is_union()); + assert_field_type(root, "contact_type", TypeKind::UnionType); + assert_field_type(root, "contact", TypeKind::Union); + let email_contact = contact + .value("EmailContact") + .and_then(EnumValueDescriptor::union_type) + .and_then(TypeDescriptor::index) + .and_then(|index| schema.object_by_index(index)) + .expect("union object index should resolve"); + assert_eq!(email_contact.name(), "test.types.EmailContact"); + } + + #[test] + fn given_reflection_base_types_when_described_should_preserve_every_type() { + let base_types = [ + (BaseType::None, TypeKind::None), + (BaseType::UType, TypeKind::UnionType), + (BaseType::Bool, TypeKind::Bool), + (BaseType::Byte, TypeKind::Int8), + (BaseType::UByte, TypeKind::Uint8), + (BaseType::Short, TypeKind::Int16), + (BaseType::UShort, TypeKind::Uint16), + (BaseType::Int, TypeKind::Int32), + (BaseType::UInt, TypeKind::Uint32), + (BaseType::Long, TypeKind::Int64), + (BaseType::ULong, TypeKind::Uint64), + (BaseType::Float, TypeKind::Float32), + (BaseType::Double, TypeKind::Float64), + (BaseType::String, TypeKind::String), + (BaseType::Vector, TypeKind::Vector), + (BaseType::Obj, TypeKind::Object), + (BaseType::Union, TypeKind::Union), + (BaseType::Array, TypeKind::Array), + (BaseType::Vector64, TypeKind::Vector64), + (BaseType(127), TypeKind::Unknown(127)), + ]; + + for (reflection_type, expected_type) in base_types { + assert_eq!(TypeKind::from(reflection_type), expected_type); + } + } + + #[test] + fn given_configured_root_table_when_loaded_should_use_override() { + let bytes = build_bfbs( + &[("example.User", false), ("example.Address", false)], + 0, + None, + ); + + let schema = FlatBufferSchema::from_bytes(bytes, Some("example.Address")) + .expect("configured root table should load"); + + assert_eq!(schema.root_table().name(), "example.Address"); + assert_eq!(schema.root_table().index(), 1); + } + + #[test] + fn given_invalid_bytes_when_loaded_should_reject_schema() { + let error = FlatBufferSchema::from_bytes(vec![1, 2, 3, 4], None) + .expect_err("invalid BFBS should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_malformed_bfbs_when_loaded_should_reject_schema() { + let mut bytes = vec![0; SIZE_UOFFSET + FILE_IDENTIFIER_LENGTH]; + bytes[SIZE_UOFFSET..].copy_from_slice(b"BFBS"); + + let error = FlatBufferSchema::from_bytes(bytes, None) + .expect_err("malformed BFBS should fail verification"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_flatbuffer_without_bfbs_identifier_when_loaded_should_reject_schema() { + let bytes = build_schema(false, &[("example.User", false)], Some(0), None); + + let error = FlatBufferSchema::from_bytes(bytes, None) + .expect_err("schema without BFBS identifier should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_missing_root_table_when_loaded_should_reject_schema() { + let bytes = build_schema(true, &[("example.User", false)], None, None); + + let error = FlatBufferSchema::from_bytes(bytes, None) + .expect_err("schema without a root table should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_unknown_configured_root_when_loaded_should_reject_schema() { + let bytes = build_bfbs(&[("example.User", false)], 0, None); + + let error = FlatBufferSchema::from_bytes(bytes, Some("example.Missing")) + .expect_err("unknown root table should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_empty_configured_root_when_loaded_should_reject_schema() { + let bytes = build_bfbs(&[("example.User", false)], 0, None); + + let error = FlatBufferSchema::from_bytes(bytes, Some(" ")) + .expect_err("empty root table should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_struct_as_configured_root_when_loaded_should_reject_schema() { + let bytes = build_bfbs(&[("example.User", false), ("example.Point", true)], 0, None); + + let error = FlatBufferSchema::from_bytes(bytes, Some("example.Point")) + .expect_err("struct cannot be selected as root table"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_invalid_file_identifier_when_loaded_should_reject_schema() { + let bytes = build_bfbs(&[("example.User", false)], 0, Some("TOO_LONG")); + + let error = FlatBufferSchema::from_bytes(bytes, None) + .expect_err("invalid file identifier should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_unknown_advanced_feature_when_loaded_should_reject_schema() { + let bytes = build_schema_with_features( + &[("example.User", false)], + Some(0), + None, + AdvancedFeatures::from_bits_retain(1 << 63), + ); + + let error = FlatBufferSchema::from_bytes(bytes, None) + .expect_err("unknown advanced schema feature should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + #[test] + fn given_missing_object_reference_when_loaded_should_reject_schema() { + let bytes = build_schema_with_field_type(TypeArgs { + base_type: BaseType::Obj, + index: 99, + ..TypeArgs::default() + }); + + let error = FlatBufferSchema::from_bytes(bytes, None) + .expect_err("missing object reference should fail"); + + assert!(matches!(error, Error::InvalidConfigValue(_))); + } + + fn build_bfbs( + objects: &[(&str, bool)], + root_index: usize, + file_identifier: Option<&str>, + ) -> Vec { + build_schema(true, objects, Some(root_index), file_identifier) + } + + fn assert_field_type(object: &ObjectDescriptor, field_name: &str, expected_type: TypeKind) { + let field = object.field(field_name).expect("field should exist"); + assert_eq!(field.field_type().base_type(), expected_type); + } + + fn build_schema( + include_bfbs_identifier: bool, + objects: &[(&str, bool)], + root_index: Option, + file_identifier: Option<&str>, + ) -> Vec { + build_schema_internal( + include_bfbs_identifier, + objects, + root_index, + file_identifier, + AdvancedFeatures::default(), + ) + } + + fn build_schema_with_features( + objects: &[(&str, bool)], + root_index: Option, + file_identifier: Option<&str>, + advanced_features: AdvancedFeatures, + ) -> Vec { + build_schema_internal( + true, + objects, + root_index, + file_identifier, + advanced_features, + ) + } + + fn build_schema_internal( + include_bfbs_identifier: bool, + objects: &[(&str, bool)], + root_index: Option, + file_identifier: Option<&str>, + advanced_features: AdvancedFeatures, + ) -> Vec { + let mut builder = FlatBufferBuilder::new(); + let empty_fields = builder.create_vector(&[] as &[WIPOffset<_>]); + let mut object_offsets = Vec::with_capacity(objects.len()); + + for (name, is_struct) in objects { + let name = builder.create_string(name); + let object = Object::create( + &mut builder, + &ObjectArgs { + name: Some(name), + fields: Some(empty_fields), + is_struct: *is_struct, + ..ObjectArgs::default() + }, + ); + object_offsets.push(object); + } + + let objects = builder.create_vector(&object_offsets); + let empty_enums = builder.create_vector(&[] as &[WIPOffset>]); + let file_identifier = file_identifier.map(|value| builder.create_string(value)); + let schema = Schema::create( + &mut builder, + &SchemaArgs { + objects: Some(objects), + enums: Some(empty_enums), + file_ident: file_identifier, + root_table: root_index.map(|index| object_offsets[index]), + advanced_features, + ..SchemaArgs::default() + }, + ); + + if include_bfbs_identifier { + finish_schema_buffer(&mut builder, schema); + } else { + builder.finish_minimal(schema); + } + + builder.finished_data().to_vec() + } + + fn build_schema_with_field_type(type_args: TypeArgs) -> Vec { + let mut builder = FlatBufferBuilder::new(); + let field_type = Type::create(&mut builder, &type_args); + let field_name = builder.create_string("child"); + let field = Field::create( + &mut builder, + &FieldArgs { + name: Some(field_name), + type_: Some(field_type), + offset: 4, + ..FieldArgs::default() + }, + ); + let fields = builder.create_vector(&[field]); + let object_name = builder.create_string("example.Root"); + let object = Object::create( + &mut builder, + &ObjectArgs { + name: Some(object_name), + fields: Some(fields), + ..ObjectArgs::default() + }, + ); + let objects = builder.create_vector(&[object]); + let enums = builder.create_vector(&[] as &[WIPOffset>]); + let schema = Schema::create( + &mut builder, + &SchemaArgs { + objects: Some(objects), + enums: Some(enums), + root_table: Some(object), + ..SchemaArgs::default() + }, + ); + finish_schema_buffer(&mut builder, schema); + + builder.finished_data().to_vec() + } +} diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index c8ed2ff94d..d5175d8370 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -39,6 +39,7 @@ pub mod api; pub mod convert; pub mod decoders; pub mod encoders; +pub mod flatbuffer; pub mod log; pub mod retry; pub mod sink; diff --git a/core/connectors/sdk/tests/fixtures/flatbuffer/types.bfbs b/core/connectors/sdk/tests/fixtures/flatbuffer/types.bfbs new file mode 100644 index 0000000000000000000000000000000000000000..2fa1ce35a51385a2e32a715b1a8ae49d1d16f675 GIT binary patch literal 2312 zcmZuyJ5Ll*6g~s{5*Ao-qsDB^#B2-&5mZ7fD8yGRjFO1SnwXGXTy^6(L)ZZ&p|r4| zr0^$DSXfw4Y-wR(L1|&(53sN>(fAs_@1D7{7x5-%?(2N#p7Wh^@Ax8e>Dr~+=%gx> zl9Q5@rGoKs@uh}dZc=RDP)!@@#Qq*oBx zLyu)+r}r&MK4RXtISw!APZ1umRb|}t;?1DI2%W#%!HmRh@QT?F~0$KKCoEMSv-5jyaNd!6Oh9Y zC+6F6GhXXrtqf*!CMxm^TM%+=ke?hCHPBC6+b!Hq&!Q#Wyvewknw|vZwb+#t^cx1sY`~>Vz@+UWy-) zll$-mSfDR`{AgBtT<@CV0$FH$Ie1D*hB$Rv(cDEWcx8;-{L{WR%|BQ*_UT(|16e^| zJz%$J@_{XHSWcKRbMy75;eLccMj#J@ zs2R`BWS}iz4$0B%4BCL&@IL+X=Q8~s_Dj&VXWZ8_hU^*iYYmh;N5~goin#3-^HMQ~ zEM^!O7vkf+DS&6@m~nqrpGr1v+IL)%1?cBdtyaIL@0F?H)OM!6i9Nm%)|a~0MFCgE zaAf^IJDY_puYnmv&Ml^)n7qZLXJn91&>ujRSpD>VWOYS-#&97$q5?iVDP;P)YJZUN zQdn({I%cEL)whN`@t%^YX?*!+5yQsY)vvznehj#FTTv9)NT1_e`b-X}Q^$w>9OM|p zc|-R7HflJ>dIL45<@H{8xhyxkntM~jd_%*vC$09X-U0XTK<}AP*Pm_7>CX)0AAk1! zj;aUz$s1eZ_;FD=Hh%Cfw)sCpUxmkvg@0$BPY<*gkNo-&)+B3}J##rVr9a;Xt*-gk zG_W=S;d|tAmOe$Wu6*T?bLG0v&jYjxE}dhx~>-TsZ!Un%sL38KCuk> zZLN(_1t(1ZCH622jLSHSKvrsHbbhGNVRpsOEpfHaf$=nFTyY#9Q(Dv^58&;Cm`^&bPo11{#@L$8>FU(MSRR|{ucnNn*b@&NJ3lpFv6 literal 0 HcmV?d00001 diff --git a/licenserc.toml b/licenserc.toml index 350f449b66..61d87eb925 100644 --- a/licenserc.toml +++ b/licenserc.toml @@ -71,6 +71,7 @@ excludes = [ "**/*.pfx", "**/*.json", "**/*.svg", + "core/connectors/sdk/tests/fixtures/flatbuffer/*.bfbs", "**/go.mod", "**/go.sum", "**/.env", From 05cdf14a609bb8cb6a1ac45b76397a6b9f50ecce Mon Sep 17 00:00:00 2001 From: StandingMan Date: Thu, 23 Jul 2026 13:26:34 +0800 Subject: [PATCH 2/2] feat(connectors): add structured payload adapters Signed-off-by: StandingMan --- .../connectors/sdk/src/encoders/flatbuffer.rs | 467 +++++++++++++----- core/connectors/sdk/src/lib.rs | 1 + core/connectors/sdk/src/structured.rs | 462 +++++++++++++++++ 3 files changed, 803 insertions(+), 127 deletions(-) create mode 100644 core/connectors/sdk/src/structured.rs diff --git a/core/connectors/sdk/src/encoders/flatbuffer.rs b/core/connectors/sdk/src/encoders/flatbuffer.rs index 32c60daaf2..b16afc42fa 100644 --- a/core/connectors/sdk/src/encoders/flatbuffer.rs +++ b/core/connectors/sdk/src/encoders/flatbuffer.rs @@ -15,12 +15,15 @@ // specific language governing permissions and limitations // under the License. -use crate::{Error, Payload, Schema, StreamEncoder}; +use std::collections::HashMap; +use std::path::PathBuf; + use base64::Engine; use flatbuffers::FlatBufferBuilder; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::path::PathBuf; + +use crate::structured::{self, StructuredValue}; +use crate::{Error, Payload, Schema, StreamEncoder}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct FlatBufferEncoderConfig { @@ -63,83 +66,57 @@ impl FlatBufferStreamEncoder { Ok(()) } - fn apply_field_transformations(&self, payload: Payload) -> Result { + fn apply_field_transformations(&self, value: StructuredValue) -> StructuredValue { 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), + match value { + StructuredValue::Object(values) => StructuredValue::Object( + values + .into_iter() + .map(|(key, value)| { + let key = mappings.get(&key).cloned().unwrap_or(key); + (key, value) + }) + .collect(), + ), + value => value, } } else { - Ok(payload) + value } } - fn encode_json_to_flatbuffer( - &self, - json_value: simd_json::OwnedValue, - ) -> Result, Error> { + fn encode_structured_to_flatbuffer(&self, value: StructuredValue) -> Result, Error> { let mut builder = FlatBufferBuilder::with_capacity(self.config.table_size_hint); - match json_value { - simd_json::OwnedValue::Object(obj) => { - let mut string_offsets = Vec::new(); - let mut string_keys = Vec::new(); + match value { + StructuredValue::Object(values) => { + let mut entries = Vec::with_capacity(values.len() * 2); - for (key, value) in obj.iter() { - let key_offset = builder.create_string(key); - string_keys.push(key_offset); + for (key, value) in values { + let key_offset = builder.create_string(&key); + entries.push(key_offset); let value_string = match value { - simd_json::OwnedValue::String(s) => s.clone(), - other => { - simd_json::to_string(other).map_err(|_| Error::InvalidJsonPayload)? - } + StructuredValue::String(value) => value, + value => self.structured_json_string(value)?, }; let value_offset = builder.create_string(&value_string); - string_offsets.push(value_offset); + entries.push(value_offset); } - let pairs: Vec<_> = string_keys.into_iter().zip(string_offsets).collect(); - - let mut vector_data = Vec::new(); - for (key_offset, value_offset) in pairs { - vector_data.push(key_offset.value()); - vector_data.push(value_offset.value()); - } - - let vector = builder.create_vector(&vector_data); + let vector = builder.create_vector(&entries); builder.finish_minimal(vector); } - simd_json::OwnedValue::Array(arr) => { - let json_string = simd_json::to_string(&simd_json::OwnedValue::Array(arr)) - .map_err(|_| Error::InvalidJsonPayload)?; - let string_offset = builder.create_string(&json_string); + StructuredValue::Bytes(data) => { + let vector = builder.create_vector(&data); + builder.finish_minimal(vector); + } + StructuredValue::String(text) => { + let string_offset = builder.create_string(&text); builder.finish_minimal(string_offset); } - other => { - let json_string = - simd_json::to_string(&other).map_err(|_| Error::InvalidJsonPayload)?; + value => { + let json_string = self.structured_json_string(value)?; let string_offset = builder.create_string(&json_string); builder.finish_minimal(string_offset); } @@ -148,27 +125,11 @@ impl FlatBufferStreamEncoder { Ok(builder.finished_data().to_vec()) } - fn encode_text_to_flatbuffer(&self, text: String) -> Result, Error> { - let mut builder = FlatBufferBuilder::with_capacity(self.config.table_size_hint); - - let mut json_bytes = text.clone().into_bytes(); - if let Ok(json_value) = simd_json::to_owned_value(&mut json_bytes) { - return self.encode_json_to_flatbuffer(json_value); - } - - let string_offset = builder.create_string(&text); - builder.finish_minimal(string_offset); - - Ok(builder.finished_data().to_vec()) - } - - fn encode_raw_to_flatbuffer(&self, data: Vec) -> Result, Error> { - let mut builder = FlatBufferBuilder::with_capacity(self.config.table_size_hint); - - let vector = builder.create_vector(&data); - builder.finish_minimal(vector); - - Ok(builder.finished_data().to_vec()) + fn structured_json_string(&self, value: StructuredValue) -> Result { + let Payload::Json(json) = structured::encode_payload(value, Schema::Json)? else { + return Err(Error::InvalidPayloadType); + }; + simd_json::to_string(&json).map_err(|_| Error::InvalidJsonPayload) } pub fn convert_format( @@ -229,15 +190,13 @@ impl StreamEncoder for FlatBufferStreamEncoder { } fn encode(&self, payload: Payload) -> Result, Error> { - let transformed_payload = self.apply_field_transformations(payload)?; - - match transformed_payload { - Payload::Json(json_value) => self.encode_json_to_flatbuffer(json_value), - Payload::Text(text) => self.encode_text_to_flatbuffer(text), - Payload::Raw(data) => self.encode_raw_to_flatbuffer(data), + match payload { Payload::FlatBuffer(data) => Ok(data), - Payload::Proto(text) => self.encode_text_to_flatbuffer(text), - Payload::Avro(data) => self.encode_raw_to_flatbuffer(data), + payload => { + let value = structured::decode_payload(payload)?; + let transformed_value = self.apply_field_transformations(value); + self.encode_structured_to_flatbuffer(transformed_value) + } } } } @@ -252,6 +211,24 @@ impl Default for FlatBufferStreamEncoder { mod tests { use super::*; + fn decode_root_string(encoded: &[u8]) -> &str { + flatbuffers::root::<&str>(encoded).unwrap() + } + + fn decode_root_bytes(encoded: &[u8]) -> Vec { + flatbuffers::root::>(encoded) + .unwrap() + .iter() + .collect() + } + + fn decode_root_strings(encoded: &[u8]) -> Vec<&str> { + flatbuffers::root::>>(encoded) + .unwrap() + .iter() + .collect() + } + #[test] fn encode_should_handle_json_payload() { let encoder = FlatBufferStreamEncoder::default(); @@ -261,36 +238,80 @@ mod tests { "age": 30 }); - let result = encoder.encode(Payload::Json(json_value)); + let encoded = encoder.encode(Payload::Json(json_value)).unwrap(); - assert!(result.is_ok()); - let encoded_data = result.unwrap(); - assert!(!encoded_data.is_empty()); + assert_eq!( + decode_root_strings(&encoded), + vec!["age", "30", "name", "John"] + ); } #[test] - fn encode_should_handle_text_payload() { + fn encode_should_preserve_plain_text_as_root_string() { let encoder = FlatBufferStreamEncoder::default(); - let text_payload = Payload::Text("Hello, World!".to_string()); - let result = encoder.encode(text_payload); + let encoded = encoder + .encode(Payload::Text("Hello, World!".to_string())) + .unwrap(); - assert!(result.is_ok()); - let encoded_data = result.unwrap(); - assert!(!encoded_data.is_empty()); + assert_eq!(decode_root_string(&encoded), "Hello, World!"); } #[test] - fn encode_should_handle_raw_payload() { + fn encode_should_preserve_raw_payload_as_root_byte_vector() { let encoder = FlatBufferStreamEncoder::default(); + let raw_data = vec![0, 1, 2, 0xff]; - let raw_data = vec![1, 2, 3, 4, 5]; - let raw_payload = Payload::Raw(raw_data); - let result = encoder.encode(raw_payload); + let encoded = encoder.encode(Payload::Raw(raw_data.clone())).unwrap(); - assert!(result.is_ok()); - let encoded_data = result.unwrap(); - assert!(!encoded_data.is_empty()); + assert_eq!(decode_root_bytes(&encoded), raw_data); + } + + #[test] + fn encode_should_preserve_empty_raw_payload() { + let encoder = FlatBufferStreamEncoder::default(); + + let encoded = encoder.encode(Payload::Raw(vec![])).unwrap(); + + assert!(decode_root_bytes(&encoded).is_empty()); + } + + #[test] + fn encode_should_serialize_structured_scalars_and_arrays_as_root_strings() { + let encoder = FlatBufferStreamEncoder::default(); + let cases = [ + (Payload::Json(simd_json::json!(null)), "null"), + (Payload::Json(simd_json::json!(true)), "true"), + (Payload::Json(simd_json::json!(-42)), "-42"), + (Payload::Json(simd_json::json!([1, 2, 3])), "[1,2,3]"), + ]; + + for (payload, expected) in cases { + let encoded = encoder.encode(payload).unwrap(); + assert_eq!(decode_root_string(&encoded), expected); + } + } + + #[test] + fn encode_should_preserve_json_string_without_quotes() { + let encoder = FlatBufferStreamEncoder::default(); + + let encoded = encoder + .encode(Payload::Json(simd_json::json!("Iggy"))) + .unwrap(); + + assert_eq!(decode_root_string(&encoded), "Iggy"); + } + + #[test] + fn encode_should_parse_json_text_before_encoding() { + let encoder = FlatBufferStreamEncoder::default(); + + let encoded = encoder + .encode(Payload::Text("[1,true,null]".to_string())) + .unwrap(); + + assert_eq!(decode_root_string(&encoded), "[1,true,null]"); } #[test] @@ -306,6 +327,20 @@ mod tests { assert_eq!(encoded_data, flatbuffer_data); } + #[test] + fn encode_should_reject_schema_dependent_payloads_without_schema() { + let encoder = FlatBufferStreamEncoder::default(); + + assert!(matches!( + encoder.encode(Payload::Proto("value".to_string())), + Err(Error::InvalidPayloadType) + )); + assert!(matches!( + encoder.encode(Payload::Avro(vec![1, 2, 3])), + Err(Error::InvalidPayloadType) + )); + } + #[test] fn encode_should_apply_field_mappings_when_configured() { let mut field_mappings = HashMap::new(); @@ -322,11 +357,64 @@ mod tests { "unchanged_field": "stays_same" }); - let result = encoder.encode(Payload::Json(json_value)); + let value = structured::decode_payload(Payload::Json(json_value)).unwrap(); + let transformed = encoder.apply_field_transformations(value); + + assert_eq!( + transformed, + StructuredValue::Object(std::collections::BTreeMap::from([ + ( + "new_field".to_string(), + StructuredValue::String("should_be_renamed".to_string()) + ), + ( + "unchanged_field".to_string(), + StructuredValue::String("stays_same".to_string()) + ), + ])) + ); + } - assert!(result.is_ok()); - let encoded_data = result.unwrap(); - assert!(!encoded_data.is_empty()); + #[test] + fn field_mappings_should_only_rename_top_level_fields() { + let mut field_mappings = HashMap::new(); + field_mappings.insert("old_field".to_string(), "new_field".to_string()); + let encoder = FlatBufferStreamEncoder::new(FlatBufferEncoderConfig { + field_mappings: Some(field_mappings), + ..FlatBufferEncoderConfig::default() + }); + let value = StructuredValue::Object(std::collections::BTreeMap::from([( + "nested".to_string(), + StructuredValue::Object(std::collections::BTreeMap::from([( + "old_field".to_string(), + StructuredValue::String("value".to_string()), + )])), + )])); + + let transformed = encoder.apply_field_transformations(value.clone()); + + assert_eq!(transformed, value); + } + + #[test] + fn field_mappings_should_leave_non_object_values_unchanged() { + let encoder = FlatBufferStreamEncoder::new(FlatBufferEncoderConfig { + field_mappings: Some(HashMap::from([( + "old_field".to_string(), + "new_field".to_string(), + )])), + ..FlatBufferEncoderConfig::default() + }); + let values = [ + StructuredValue::Null, + StructuredValue::String("value".to_string()), + StructuredValue::Array(vec![StructuredValue::U64(1)]), + StructuredValue::Bytes(vec![1, 2, 3]), + ]; + + for value in values { + assert_eq!(encoder.apply_field_transformations(value.clone()), value); + } } #[test] @@ -337,17 +425,12 @@ mod tests { let result = encoder.convert_format(Payload::FlatBuffer(flatbuffer_data.clone()), Schema::Json); - 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("flatbuffer_size")); - assert!(map.contains_key("raw_data_base64")); - } else { - panic!("Expected JSON object"); - } - } else { - panic!("Expected JSON payload"); - } + assert!( + matches!(result, Ok(Payload::Json(value)) if value == simd_json::json!({ + "flatbuffer_size": 5, + "raw_data_base64": "AQIDBAU=" + })) + ); } #[test] @@ -357,14 +440,111 @@ mod tests { let flatbuffer_data = vec![1, 2, 3, 4, 5]; let result = encoder.convert_format(Payload::FlatBuffer(flatbuffer_data), Schema::Text); - assert!(result.is_ok()); - if let Ok(Payload::Text(text)) = result { - assert!(!text.is_empty()); - } else { - panic!("Expected Text payload"); + assert!(matches!(result, Ok(Payload::Text(text)) if text == "AQIDBAU=")); + } + + #[test] + fn convert_format_should_transform_flatbuffer_to_raw_without_modification() { + let encoder = FlatBufferStreamEncoder::default(); + let data = vec![0, 1, 2, 0xff]; + + let result = encoder.convert_format(Payload::FlatBuffer(data.clone()), Schema::Raw); + + assert!(matches!(result, Ok(Payload::Raw(bytes)) if bytes == data)); + } + + #[test] + fn convert_format_should_transform_json_to_text_and_raw() { + let encoder = FlatBufferStreamEncoder::default(); + let value = simd_json::json!({"name": "Iggy", "active": true}); + + let text = encoder + .convert_format(Payload::Json(value.clone()), Schema::Text) + .unwrap(); + let raw = encoder + .convert_format(Payload::Json(value.clone()), Schema::Raw) + .unwrap(); + + let Payload::Text(text) = text else { + panic!("expected text payload"); + }; + let Payload::Raw(mut raw) = raw else { + panic!("expected raw payload"); + }; + let mut text = text.into_bytes(); + assert_eq!(simd_json::to_owned_value(&mut text).unwrap(), value); + assert_eq!(simd_json::to_owned_value(&mut raw).unwrap(), value); + } + + #[test] + fn convert_format_should_transform_valid_text_and_raw_json_to_json() { + let encoder = FlatBufferStreamEncoder::default(); + let expected = simd_json::json!({"name": "Iggy"}); + + for payload in [ + Payload::Text(r#"{"name":"Iggy"}"#.to_string()), + Payload::Raw(br#"{"name":"Iggy"}"#.to_vec()), + ] { + assert!(matches!( + encoder.convert_format(payload, Schema::Json), + Ok(Payload::Json(value)) if value == expected + )); } } + #[test] + fn convert_format_should_reject_invalid_json_text_and_raw() { + let encoder = FlatBufferStreamEncoder::default(); + + for payload in [ + Payload::Text("not json".to_string()), + Payload::Raw(b"not json".to_vec()), + ] { + assert_eq!( + encoder.convert_format(payload, Schema::Json).unwrap_err(), + Error::InvalidJsonPayload + ); + } + } + + #[test] + fn convert_format_should_transform_text_and_raw_utf8_bidirectionally() { + let encoder = FlatBufferStreamEncoder::default(); + let text = "你好, Iggy"; + + assert!(matches!( + encoder.convert_format(Payload::Text(text.to_string()), Schema::Raw), + Ok(Payload::Raw(bytes)) if bytes == text.as_bytes() + )); + assert!(matches!( + encoder.convert_format(Payload::Raw(text.as_bytes().to_vec()), Schema::Text), + Ok(Payload::Text(value)) if value == text + )); + } + + #[test] + fn convert_format_should_reject_non_utf8_raw_as_text() { + let encoder = FlatBufferStreamEncoder::default(); + + let result = encoder.convert_format(Payload::Raw(vec![0xff, 0xfe]), Schema::Text); + + assert_eq!(result.unwrap_err(), Error::InvalidTextPayload); + } + + #[test] + fn convert_format_should_pass_through_same_format() { + let encoder = FlatBufferStreamEncoder::default(); + + assert!(matches!( + encoder.convert_format(Payload::Text("value".to_string()), Schema::Text), + Ok(Payload::Text(value)) if value == "value" + )); + assert!(matches!( + encoder.convert_format(Payload::Raw(vec![1, 2, 3]), Schema::Raw), + Ok(Payload::Raw(value)) if value == vec![1, 2, 3] + )); + } + #[test] fn config_should_have_sensible_defaults() { let config = FlatBufferEncoderConfig::default(); @@ -392,4 +572,37 @@ mod tests { assert_eq!(encoder.config.root_table_name, config.root_table_name); assert_eq!(encoder.config.table_size_hint, config.table_size_hint); } + + #[test] + fn update_config_should_replace_the_complete_configuration() { + let mut encoder = FlatBufferStreamEncoder::default(); + let config = FlatBufferEncoderConfig { + schema_path: Some(PathBuf::from("schema.bfbs")), + root_table_name: Some("Root".to_string()), + field_mappings: Some(HashMap::from([("old".to_string(), "new".to_string())])), + preserve_unknown_fields: true, + include_paths: vec![PathBuf::from("schemas"), PathBuf::from("includes")], + table_size_hint: 4096, + }; + + encoder.update_config(config.clone()).unwrap(); + + assert_eq!(encoder.config.schema_path, config.schema_path); + assert_eq!(encoder.config.root_table_name, config.root_table_name); + assert_eq!(encoder.config.field_mappings, config.field_mappings); + assert_eq!( + encoder.config.preserve_unknown_fields, + config.preserve_unknown_fields + ); + assert_eq!(encoder.config.include_paths, config.include_paths); + assert_eq!(encoder.config.table_size_hint, config.table_size_hint); + } + + #[test] + fn encoder_should_report_flatbuffer_schema() { + assert_eq!( + FlatBufferStreamEncoder::default().schema(), + Schema::FlatBuffer + ); + } } diff --git a/core/connectors/sdk/src/lib.rs b/core/connectors/sdk/src/lib.rs index d5175d8370..d43a21a4d7 100644 --- a/core/connectors/sdk/src/lib.rs +++ b/core/connectors/sdk/src/lib.rs @@ -44,6 +44,7 @@ pub mod log; pub mod retry; pub mod sink; pub mod source; +pub(crate) mod structured; pub mod transforms; pub use convert::owned_value_to_serde_json; diff --git a/core/connectors/sdk/src/structured.rs b/core/connectors/sdk/src/structured.rs new file mode 100644 index 0000000000..35fd26841a --- /dev/null +++ b/core/connectors/sdk/src/structured.rs @@ -0,0 +1,462 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::collections::BTreeMap; + +use crate::{Error, Payload, Schema}; + +#[derive(Debug, Clone, PartialEq)] +pub(crate) enum StructuredValue { + Null, + Bool(bool), + I64(i64), + U64(u64), + F64(f64), + String(String), + Bytes(Vec), + Array(Vec), + Object(BTreeMap), +} + +pub(crate) trait FormatAdapter: Send + Sync { + fn decode_payload(&self, payload: Payload) -> Result; + fn encode_payload(&self, value: StructuredValue) -> Result; +} + +struct JsonAdapter; +struct TextAdapter; +struct RawAdapter; + +static JSON_ADAPTER: JsonAdapter = JsonAdapter; +static TEXT_ADAPTER: TextAdapter = TextAdapter; +static RAW_ADAPTER: RawAdapter = RawAdapter; + +pub(crate) fn decode_payload(payload: Payload) -> Result { + let adapter = match &payload { + Payload::Json(_) => adapter_for_schema(Schema::Json)?, + Payload::Text(_) => adapter_for_schema(Schema::Text)?, + Payload::Raw(_) => adapter_for_schema(Schema::Raw)?, + Payload::Proto(_) | Payload::FlatBuffer(_) | Payload::Avro(_) => { + return Err(Error::InvalidPayloadType); + } + }; + + adapter.decode_payload(payload) +} + +pub(crate) fn encode_payload( + value: StructuredValue, + target_schema: Schema, +) -> Result { + adapter_for_schema(target_schema)?.encode_payload(value) +} + +fn adapter_for_schema(schema: Schema) -> Result<&'static dyn FormatAdapter, Error> { + match schema { + Schema::Json => Ok(&JSON_ADAPTER), + Schema::Text => Ok(&TEXT_ADAPTER), + Schema::Raw => Ok(&RAW_ADAPTER), + Schema::Proto | Schema::FlatBuffer | Schema::Avro => Err(Error::InvalidPayloadType), + } +} + +impl FormatAdapter for JsonAdapter { + fn decode_payload(&self, payload: Payload) -> Result { + match payload { + Payload::Json(value) => Ok(value.into()), + _ => Err(Error::InvalidPayloadType), + } + } + + fn encode_payload(&self, value: StructuredValue) -> Result { + Ok(Payload::Json(value.try_into()?)) + } +} + +impl FormatAdapter for TextAdapter { + fn decode_payload(&self, payload: Payload) -> Result { + let Payload::Text(text) = payload else { + return Err(Error::InvalidPayloadType); + }; + + let mut bytes = text.clone().into_bytes(); + match simd_json::to_owned_value(&mut bytes) { + Ok(value) => Ok(value.into()), + Err(_) => Ok(StructuredValue::String(text)), + } + } + + fn encode_payload(&self, value: StructuredValue) -> Result { + match value { + StructuredValue::String(text) => Ok(Payload::Text(text)), + StructuredValue::Bytes(bytes) => String::from_utf8(bytes) + .map(Payload::Text) + .map_err(|_| Error::InvalidTextPayload), + value => structured_json_string(value).map(Payload::Text), + } + } +} + +impl FormatAdapter for RawAdapter { + fn decode_payload(&self, payload: Payload) -> Result { + match payload { + Payload::Raw(bytes) => Ok(StructuredValue::Bytes(bytes)), + _ => Err(Error::InvalidPayloadType), + } + } + + fn encode_payload(&self, value: StructuredValue) -> Result { + match value { + StructuredValue::Bytes(bytes) => Ok(Payload::Raw(bytes)), + StructuredValue::String(text) => Ok(Payload::Raw(text.into_bytes())), + value => structured_json_bytes(value).map(Payload::Raw), + } + } +} + +impl From for StructuredValue { + fn from(value: simd_json::OwnedValue) -> Self { + match value { + simd_json::OwnedValue::Static(node) => match node { + simd_json::StaticNode::Null => Self::Null, + simd_json::StaticNode::Bool(value) => Self::Bool(value), + simd_json::StaticNode::I64(value) => Self::I64(value), + simd_json::StaticNode::U64(value) => Self::U64(value), + simd_json::StaticNode::F64(value) => Self::F64(value), + }, + simd_json::OwnedValue::String(value) => Self::String(value), + simd_json::OwnedValue::Array(values) => { + Self::Array(values.into_iter().map(Self::from).collect()) + } + simd_json::OwnedValue::Object(values) => Self::Object( + values + .into_iter() + .map(|(key, value)| (key, Self::from(value))) + .collect(), + ), + } + } +} + +impl TryFrom for simd_json::OwnedValue { + type Error = Error; + + fn try_from(value: StructuredValue) -> Result { + match value { + StructuredValue::Null => Ok(Self::Static(simd_json::StaticNode::Null)), + StructuredValue::Bool(value) => Ok(Self::Static(simd_json::StaticNode::Bool(value))), + StructuredValue::I64(value) => Ok(Self::Static(simd_json::StaticNode::I64(value))), + StructuredValue::U64(value) => Ok(Self::Static(simd_json::StaticNode::U64(value))), + StructuredValue::F64(value) => Ok(Self::Static(simd_json::StaticNode::F64(value))), + StructuredValue::String(value) => Ok(Self::String(value)), + StructuredValue::Bytes(_) => Err(Error::InvalidRecordValue( + "binary structured value cannot be represented as JSON".to_string(), + )), + StructuredValue::Array(values) => values + .into_iter() + .map(Self::try_from) + .collect::, _>>() + .map(Box::new) + .map(Self::Array), + StructuredValue::Object(values) => { + let object: Result = values + .into_iter() + .map(|(key, value)| Self::try_from(value).map(|value| (key, value))) + .collect(); + Ok(Self::Object(Box::new(object?))) + } + } + } +} + +fn structured_json_string(value: StructuredValue) -> Result { + let json: simd_json::OwnedValue = value.try_into()?; + simd_json::to_string(&json).map_err(|_| Error::InvalidJsonPayload) +} + +fn structured_json_bytes(value: StructuredValue) -> Result, Error> { + let json: simd_json::OwnedValue = value.try_into()?; + simd_json::to_vec(&json).map_err(|_| Error::InvalidJsonPayload) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn nested_value() -> StructuredValue { + StructuredValue::Object(BTreeMap::from([ + ( + "array".to_string(), + StructuredValue::Array(vec![ + StructuredValue::Null, + StructuredValue::Bool(true), + StructuredValue::I64(-42), + StructuredValue::U64(42), + StructuredValue::F64(1.5), + StructuredValue::String("value".to_string()), + ]), + ), + ( + "object".to_string(), + StructuredValue::Object(BTreeMap::from([( + "empty".to_string(), + StructuredValue::Array(vec![]), + )])), + ), + ])) + } + + #[test] + fn json_adapter_should_round_trip_all_json_value_kinds() { + let original = simd_json::json!({ + "null": null, + "bool": true, + "i64": -1, + "u64": u64::MAX, + "f64": 1.5, + "string": "value", + "array": [1, 2, 3] + }); + + let value = decode_payload(Payload::Json(original.clone())).unwrap(); + let encoded = encode_payload(value, Schema::Json).unwrap(); + + assert!(matches!(encoded, Payload::Json(value) if value == original)); + } + + #[test] + fn json_conversion_should_preserve_numeric_variants() { + let cases = [ + ( + simd_json::OwnedValue::Static(simd_json::StaticNode::I64(-1)), + StructuredValue::I64(-1), + ), + ( + simd_json::OwnedValue::Static(simd_json::StaticNode::U64(u64::MAX)), + StructuredValue::U64(u64::MAX), + ), + ( + simd_json::OwnedValue::Static(simd_json::StaticNode::F64(1.25)), + StructuredValue::F64(1.25), + ), + ]; + + for (json, expected) in cases { + assert_eq!(StructuredValue::from(json), expected); + } + } + + #[test] + fn json_adapter_should_round_trip_nested_structures() { + let original = nested_value(); + + let encoded = encode_payload(original.clone(), Schema::Json).unwrap(); + let decoded = decode_payload(encoded).unwrap(); + + assert_eq!(decoded, original); + } + + #[test] + fn text_adapter_should_parse_json_objects() { + let value = decode_payload(Payload::Text(r#"{"name":"Iggy"}"#.to_string())).unwrap(); + + assert!(matches!(value, StructuredValue::Object(_))); + } + + #[test] + fn text_adapter_should_parse_each_json_root_kind() { + let cases = [ + ("null", StructuredValue::Null), + ("true", StructuredValue::Bool(true)), + ("-42", StructuredValue::I64(-42)), + ("42", StructuredValue::U64(42)), + ("1.5", StructuredValue::F64(1.5)), + (r#""value""#, StructuredValue::String("value".to_string())), + ( + "[1,2]", + StructuredValue::Array(vec![StructuredValue::U64(1), StructuredValue::U64(2)]), + ), + ]; + + for (text, expected) in cases { + assert_eq!( + decode_payload(Payload::Text(text.to_string())).unwrap(), + expected + ); + } + } + + #[test] + fn text_adapter_should_preserve_plain_text() { + let value = decode_payload(Payload::Text("plain text".to_string())).unwrap(); + let encoded = encode_payload(value, Schema::Text).unwrap(); + + assert!(matches!(encoded, Payload::Text(text) if text == "plain text")); + } + + #[test] + fn text_adapter_should_preserve_empty_and_whitespace_only_text() { + for text in ["", " ", "\n\t"] { + assert_eq!( + decode_payload(Payload::Text(text.to_string())).unwrap(), + StructuredValue::String(text.to_string()) + ); + } + } + + #[test] + fn text_adapter_should_encode_nested_structures_as_json() { + let encoded = encode_payload(nested_value(), Schema::Text).unwrap(); + let Payload::Text(text) = encoded else { + panic!("expected text payload"); + }; + let mut bytes = text.into_bytes(); + let json = simd_json::to_owned_value(&mut bytes).unwrap(); + + assert_eq!( + StructuredValue::from(json), + nested_value(), + "encoded text must preserve the complete structured value" + ); + } + + #[test] + fn text_adapter_should_decode_valid_utf8_bytes() { + let encoded = encode_payload( + StructuredValue::Bytes("你好, Iggy".as_bytes().to_vec()), + Schema::Text, + ) + .unwrap(); + + assert!(matches!(encoded, Payload::Text(text) if text == "你好, Iggy")); + } + + #[test] + fn text_adapter_should_reject_non_utf8_bytes() { + let result = encode_payload(StructuredValue::Bytes(vec![0xff, 0xfe]), Schema::Text); + + assert_eq!(result.unwrap_err(), Error::InvalidTextPayload); + } + + #[test] + fn raw_adapter_should_preserve_non_utf8_bytes() { + let bytes = vec![0, 159, 146, 150]; + let value = decode_payload(Payload::Raw(bytes.clone())).unwrap(); + let encoded = encode_payload(value, Schema::Raw).unwrap(); + + assert!(matches!(encoded, Payload::Raw(encoded) if encoded == bytes)); + } + + #[test] + fn raw_adapter_should_encode_strings_without_json_quotes() { + let encoded = encode_payload( + StructuredValue::String("plain text".to_string()), + Schema::Raw, + ) + .unwrap(); + + assert!(matches!(encoded, Payload::Raw(bytes) if bytes == b"plain text")); + } + + #[test] + fn raw_adapter_should_encode_nested_structures_as_json_bytes() { + let encoded = encode_payload(nested_value(), Schema::Raw).unwrap(); + let Payload::Raw(mut bytes) = encoded else { + panic!("expected raw payload"); + }; + let json = simd_json::to_owned_value(&mut bytes).unwrap(); + + assert_eq!(StructuredValue::from(json), nested_value()); + } + + #[test] + fn json_adapter_should_reject_binary_values() { + let result = encode_payload(StructuredValue::Bytes(vec![1, 2, 3]), Schema::Json); + + assert!(matches!(result, Err(Error::InvalidRecordValue(_)))); + } + + #[test] + fn structured_encoders_should_reject_nested_binary_values() { + let value = StructuredValue::Object(BTreeMap::from([( + "binary".to_string(), + StructuredValue::Bytes(vec![1, 2, 3]), + )])); + + for schema in [Schema::Json, Schema::Text, Schema::Raw] { + assert!( + matches!( + encode_payload(value.clone(), schema), + Err(Error::InvalidRecordValue(_)) + ), + "{schema} must not silently discard nested binary data" + ); + } + } + + #[test] + fn decode_payload_should_reject_schema_dependent_formats() { + let payloads = [ + Payload::Proto("value".to_string()), + Payload::FlatBuffer(vec![1, 2, 3]), + Payload::Avro(vec![1, 2, 3]), + ]; + + for payload in payloads { + assert_eq!( + decode_payload(payload).unwrap_err(), + Error::InvalidPayloadType + ); + } + } + + #[test] + fn adapter_factory_should_reject_schema_dependent_formats() { + for schema in [Schema::Proto, Schema::FlatBuffer, Schema::Avro] { + assert!(matches!( + adapter_for_schema(schema), + Err(Error::InvalidPayloadType) + )); + assert_eq!( + encode_payload(StructuredValue::Null, schema).unwrap_err(), + Error::InvalidPayloadType + ); + } + } + + #[test] + fn concrete_adapters_should_reject_the_wrong_payload_variant() { + assert_eq!( + JSON_ADAPTER + .decode_payload(Payload::Text("value".to_string())) + .unwrap_err(), + Error::InvalidPayloadType + ); + assert_eq!( + TEXT_ADAPTER + .decode_payload(Payload::Raw(vec![1])) + .unwrap_err(), + Error::InvalidPayloadType + ); + assert_eq!( + RAW_ADAPTER + .decode_payload(Payload::Json(simd_json::json!(null))) + .unwrap_err(), + Error::InvalidPayloadType + ); + } +}