fix

Author: callicles

apps/framework-cli/src/utilities/validate_passthrough.rs

@@ -5,7 +5,7 @@ use crate::infrastructure::processes::kafka_clickhouse_sync::IPV4_PATTERN;
 use itertools::Either;
 use num_bigint::{BigInt, BigUint};
 use regex::Regex;
-use serde::de::{DeserializeSeed, Error, MapAccess, SeqAccess, Visitor};
+use serde::de::{Deserialize, DeserializeSeed, Error, MapAccess, SeqAccess, Visitor};
 use serde::ser::{SerializeMap, SerializeSeq};
 use serde::{forward_to_deserialize_any, Deserializer, Serialize, Serializer};
 use serde_json::Value;
@@ -616,11 +616,10 @@ impl<'de, S: SerializeValue> Visitor<'de> for &mut ValueVisitor<'_, S> {
     {
         match self.t {
             ColumnType::Nested(ref fields) => {
-                let inner = DataModelVisitor::with_context(
+                let inner = DataModelVisitor::with_nested_context(
                     &fields.columns,
-                    Some(&self.context),
+                    &self.context,
                     self.jwt_claims,
-                    false, // Nested types don't allow extra fields
                 );
                 let serializer = MapAccessSerializer {
                     inner: RefCell::new(inner),
@@ -654,12 +653,8 @@ impl<'de, S: SerializeValue> Visitor<'de> for &mut ValueVisitor<'_, S> {
                         }
                     })
                     .collect();
-                let inner = DataModelVisitor::with_context(
-                    &columns,
-                    Some(&self.context),
-                    self.jwt_claims,
-                    false, // NamedTuples don't allow extra fields
-                );
+                let inner =
+                    DataModelVisitor::with_nested_context(&columns, &self.context, self.jwt_claims);
                 let serializer = MapAccessSerializer {
                     inner: RefCell::new(inner),
                     map: RefCell::new(map),
@@ -1008,6 +1003,120 @@ impl<'de, A: MapAccess<'de>> Serialize for MapAccessSerializer<'de, '_, A> {
     }
 }
 
+/// Helper seed for streaming pass-through of arbitrary values without intermediate allocation.
+/// This is used to efficiently pass through extra fields in ingest payloads.
+/// For complex nested structures, falls back to serde_json::Value, but handles primitives
+/// and simple types without allocation.
+struct PassThroughSeed<'a, S: SerializeMap>(&'a mut S);
+
+impl<'de, S: SerializeMap> DeserializeSeed<'de> for PassThroughSeed<'_, S> {
+    type Value = ();
+
+    fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        // Use a visitor to deserialize and immediately serialize primitives
+        struct PassThroughVisitor<'a, S: SerializeMap>(&'a mut S);
+
+        impl<'de, S: SerializeMap> Visitor<'de> for PassThroughVisitor<'_, S> {
+            type Value = ();
+
+            fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result {
+                formatter.write_str("any valid JSON value")
+            }
+
+            fn visit_bool<E>(self, v: bool) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(&v).map_err(E::custom)
+            }
+
+            fn visit_i64<E>(self, v: i64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(&v).map_err(E::custom)
+            }
+
+            fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(&v).map_err(E::custom)
+            }
+
+            fn visit_f64<E>(self, v: f64) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(&v).map_err(E::custom)
+            }
+
+            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(v).map_err(E::custom)
+            }
+
+            fn visit_string<E>(self, v: String) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(&v).map_err(E::custom)
+            }
+
+            fn visit_none<E>(self) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0
+                    .serialize_value(&Option::<()>::None)
+                    .map_err(E::custom)
+            }
+
+            fn visit_some<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+            where
+                D: Deserializer<'de>,
+            {
+                // For Option<T>, recursively pass through the inner value
+                PassThroughSeed(self.0).deserialize(deserializer)
+            }
+
+            fn visit_unit<E>(self) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                self.0.serialize_value(&()).map_err(E::custom)
+            }
+
+            fn visit_seq<A>(self, seq: A) -> Result<Self::Value, A::Error>
+            where
+                A: SeqAccess<'de>,
+            {
+                // For complex nested structures (arrays/objects), fall back to Value
+                // to avoid complexity. In practice, extra fields are usually primitives.
+                let value = Value::deserialize(serde::de::value::SeqAccessDeserializer::new(seq))?;
+                self.0.serialize_value(&value).map_err(A::Error::custom)
+            }
+
+            fn visit_map<A>(self, map: A) -> Result<Self::Value, A::Error>
+            where
+                A: MapAccess<'de>,
+            {
+                // For complex nested structures (arrays/objects), fall back to Value
+                // to avoid complexity. In practice, extra fields are usually primitives.
+                let value = Value::deserialize(serde::de::value::MapAccessDeserializer::new(map))?;
+                self.0.serialize_value(&value).map_err(A::Error::custom)
+            }
+        }
+
+        deserializer.deserialize_any(PassThroughVisitor(self.0))
+    }
+}
+
 pub struct DataModelVisitor<'a> {
     columns: HashMap<String, (Column, State)>,
     parent_context: Option<&'a ParentContext<'a>>,
@@ -1043,6 +1152,16 @@ impl<'a> DataModelVisitor<'a> {
         }
     }
 
+    /// Create a visitor for nested structures (Nested types and NamedTuples).
+    /// These types don't allow extra fields, so `allow_extra_fields` is always false.
+    fn with_nested_context(
+        columns: &[Column],
+        parent_context: &'a ParentContext<'a>,
+        jwt_claims: Option<&'a Value>,
+    ) -> Self {
+        Self::with_context(columns, Some(parent_context), jwt_claims, false)
+    }
+
     fn transfer_map_access_to_serialize_map<'de, A: MapAccess<'de>, S: SerializeMap>(
         &mut self,
         map: &mut A,
@@ -1069,13 +1188,11 @@ impl<'a> DataModelVisitor<'a> {
                 map.next_value_seed(&mut visitor)?;
             } else if self.allow_extra_fields {
                 // Pass through extra fields when allowed (e.g., for types with index signatures)
-                let value: Value = map.next_value()?;
+                // Use streaming pass-through to avoid intermediate Value allocation
                 map_serializer
                     .serialize_key(&key)
                     .map_err(A::Error::custom)?;
-                map_serializer
-                    .serialize_value(&value)
-                    .map_err(A::Error::custom)?;
+                map.next_value_seed(PassThroughSeed(map_serializer))?;
             } else {
                 map.next_value::<serde::de::IgnoredAny>()?;
             }