feat(clickhouse): read tuple fields inside query DAGs

control_plane/src/query_plan/clickhouse_exact.rs

@@ -78,6 +78,7 @@ fn scalar(expr: &QueryExpr, schema: &Schema) -> Result<String, String> {
                 "map" => "map",
                 "mapconcat" => "mapConcat",
                 "asap_map_access" | "asap_element_access" => "arrayElement",
+                "asap_struct_field" => "tupleElement",
                 _ => return Err(format!("unsupported exact scalar function {name}")),
             };
             expr.scalar_type(schema).map_err(|e| e.to_string())?;
@@ -323,6 +324,31 @@ mod tests {
         );
     }
 
+    #[test]
+    fn typed_struct_field_renders_native_lookup() {
+        let schema = Schema::new(vec![Column::new(
+            "sample",
+            DataType::Struct {
+                fields: vec![
+                    Column::new("ts", DataType::Int64, false),
+                    Column::new("value", DataType::Float64, true),
+                ],
+            },
+            false,
+        )]);
+        let expr = QueryExpr::FunctionCall {
+            name: "asap_struct_field".into(),
+            args: vec![
+                QueryExpr::Column(0),
+                QueryExpr::Literal(ScalarValue::Utf8("value".into())),
+            ],
+        };
+        assert_eq!(
+            scalar(&expr, &schema).unwrap(),
+            "tupleElement(`sample`, 'value')"
+        );
+    }
+
     #[test]
     fn unsupported_scalar_is_not_forwarded_as_arbitrary_native_code() {
         let schema = Schema::new(vec![]);

data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs

@@ -1,6 +1,7 @@
 //! ClickHouse row semantics for planner-owned relational wrappers.
 
 mod aggregate;
+mod collection;
 
 use std::{cmp::Ordering, collections::BTreeMap, sync::Arc};
 
@@ -42,6 +43,7 @@ enum Cell {
     Timestamp(i64),
     Map(Vec<(Cell, Cell)>),
     List(Arc<[Cell]>),
+    Struct(Arc<[Cell]>),
 }
 
 fn json_cell(
@@ -75,9 +77,25 @@ fn json_cell(
                     .into(),
             ))
         }
-        DataType::Struct { .. } => Err(ClickHouseRelationalError::Unsupported(
-            "struct value transport".into(),
-        )),
+        DataType::Struct { fields } => {
+            let types =
+                collection::tuple_field_types(clickhouse_type, fields).ok_or_else(invalid)?;
+            let items = value
+                .as_array()
+                .filter(|items| items.len() == fields.len())
+                .ok_or_else(invalid)?;
+            Ok(Cell::Struct(
+                items
+                    .iter()
+                    .zip(fields)
+                    .zip(types)
+                    .map(|((item, field), native)| {
+                        json_cell(item, &field.dtype, field.nullable, native)
+                    })
+                    .collect::<Result<Vec<_>, _>>()?
+                    .into(),
+            ))
+        }
         DataType::Int64 => value.as_i64().map(Cell::Int64).ok_or_else(invalid),
         DataType::Float64 => value.as_f64().map(Cell::Float64).ok_or_else(invalid),
         DataType::Utf8 => value
@@ -289,20 +307,11 @@ impl ClickHouseRelation {
 
 fn map_type_parts(actual: &str) -> Option<(&str, &str)> {
     let inner = actual.trim().strip_prefix("Map(")?.strip_suffix(')')?;
-    let mut depth = 0_i32;
-    let mut quoted = false;
-    for (index, ch) in inner.char_indices() {
-        match ch {
-            '\'' => quoted = !quoted,
-            '(' if !quoted => depth += 1,
-            ')' if !quoted => depth -= 1,
-            ',' if !quoted && depth == 0 => {
-                return Some((inner[..index].trim(), inner[index + 1..].trim()))
-            }
-            _ => {}
-        }
-    }
-    None
+    let args = collection::arguments(inner)?;
+    let [key, value] = args.as_slice() else {
+        return None;
+    };
+    Some((*key, *value))
 }
 
 fn clickhouse_type_matches(actual: Option<&str>, expected: &DataType, nullable: bool) -> bool {
@@ -331,7 +340,9 @@ fn clickhouse_type_matches(actual: Option<&str>, expected: &DataType, nullable:
                         clickhouse_type_matches(Some(inner), &element.dtype, element.nullable)
                     })
         }
-        DataType::Struct { .. } => false,
+        DataType::Struct { fields } => {
+            !nullable && collection::tuple_field_types(actual, fields).is_some()
+        }
         DataType::Int64 => actual == "Int64",
         DataType::Float64 => actual == "Float64",
         DataType::Utf8 => actual == "String",
@@ -578,6 +589,37 @@ fn eval(
         }
         QueryExpr::FunctionCall { name, args } => {
             use planner_types::pre_asap::scalar_signature::MapScalarFunction;
+            if name.eq_ignore_ascii_case("asap_struct_field") {
+                expr.scalar_type(schema)
+                    .map_err(|error| ClickHouseRelationalError::Invalid(error.to_string()))?;
+                let DataType::Struct { fields } = args[0]
+                    .scalar_type(schema)
+                    .map_err(|error| ClickHouseRelationalError::Invalid(error.to_string()))?
+                    .0
+                else {
+                    unreachable!()
+                };
+                let offset = match &args[1] {
+                    QueryExpr::Literal(ScalarValue::Int64(index)) => {
+                        usize::try_from(index - 1).ok()
+                    }
+                    QueryExpr::Literal(ScalarValue::Utf8(name)) => {
+                        fields.iter().position(|field| &field.name == name)
+                    }
+                    _ => None,
+                }
+                .ok_or_else(|| {
+                    ClickHouseRelationalError::Invalid("struct field selector".into())
+                })?;
+                let Cell::Struct(values) = eval(&args[0], row, schema)? else {
+                    return Err(ClickHouseRelationalError::Invalid(
+                        "struct field input".into(),
+                    ));
+                };
+                return values.get(offset).cloned().ok_or_else(|| {
+                    ClickHouseRelationalError::Invalid("struct field value".into())
+                });
+            }
             if name.eq_ignore_ascii_case("asap_element_access") {
                 let (output_type, _) = expr
                     .scalar_type(schema)
@@ -717,6 +759,13 @@ fn default_collection_element(
         DataType::Bool => Cell::Bool(false),
         DataType::Map { .. } => Cell::Map(Vec::new()),
         DataType::List { .. } => Cell::List(Arc::from([])),
+        DataType::Struct { fields } => Cell::Struct(
+            fields
+                .iter()
+                .map(|field| default_collection_element(&field.dtype, field.nullable))
+                .collect::<Result<Vec<_>, _>>()?
+                .into(),
+        ),
         _ => {
             return Err(ClickHouseRelationalError::Unsupported(
                 "collection missing-element default type".into(),
@@ -845,7 +894,7 @@ fn compare_sort_keys(
 fn contains_nan(value: &Cell) -> bool {
     match value {
         Cell::Float64(value) => value.is_nan(),
-        Cell::List(values) => values.iter().any(contains_nan),
+        Cell::List(values) | Cell::Struct(values) => values.iter().any(contains_nan),
         Cell::Map(entries) => entries
             .iter()
             .any(|(key, value)| contains_nan(key) || contains_nan(value)),
@@ -1187,6 +1236,63 @@ mod tests {
         assert!(eval(&zero, &[Cell::List(Arc::from([])), Cell::Int64(0)], &schema).is_err());
     }
 
+    #[test]
+    fn nested_array_tuple_access_preserves_fields_and_defaults() {
+        use planner_types::pre_asap::{Column, Schema};
+        let tuple = DataType::Struct {
+            fields: vec![
+                Column::new("ts", DataType::Int64, false),
+                Column::new("value", DataType::Float64, true),
+            ],
+        };
+        let dtype = DataType::List {
+            element: Box::new(Column::new("item", tuple, false)),
+        };
+        let native = "Array(Tuple(ts Int64, value Nullable(Float64)))";
+        assert!(clickhouse_type_matches(Some(native), &dtype, false));
+        let samples = json_cell(
+            &serde_json::json!([[9007199254740993_i64, 2.5], [7, null]]),
+            &dtype,
+            false,
+            native,
+        )
+        .unwrap();
+        let schema = Schema::new(vec![Column::new("samples", dtype, false)]);
+        let field = |index, name: &str| QueryExpr::FunctionCall {
+            name: "asap_struct_field".into(),
+            args: vec![
+                QueryExpr::FunctionCall {
+                    name: "asap_element_access".into(),
+                    args: vec![
+                        QueryExpr::Column(0),
+                        QueryExpr::Literal(ScalarValue::Int64(index)),
+                    ],
+                },
+                QueryExpr::Literal(ScalarValue::Utf8(name.into())),
+            ],
+        };
+        assert_eq!(
+            eval(&field(1, "ts"), std::slice::from_ref(&samples), &schema).unwrap(),
+            Cell::Int64(9007199254740993)
+        );
+        assert_eq!(
+            eval(&field(1, "value"), std::slice::from_ref(&samples), &schema).unwrap(),
+            Cell::Float64(2.5)
+        );
+        assert_eq!(
+            eval(&field(-1, "value"), std::slice::from_ref(&samples), &schema).unwrap(),
+            Cell::Null
+        );
+        assert_eq!(
+            eval(&field(99, "ts"), std::slice::from_ref(&samples), &schema).unwrap(),
+            Cell::Int64(0)
+        );
+        assert_eq!(
+            eval(&field(99, "value"), std::slice::from_ref(&samples), &schema).unwrap(),
+            Cell::Null
+        );
+    }
+
     fn schema(fields: &[(&str, DataType)]) -> SummarySchema {
         SummarySchema {
             fields: fields

data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter/collection.rs

@@ -0,0 +1,120 @@
+//! Native collection metadata is checked against the existing shared schema.
+
+use planner_types::pre_asap::Column;
+
+/// Split native type arguments without splitting nested types or quoted names.
+pub(super) fn arguments(input: &str) -> Option<Vec<&str>> {
+    let mut result = Vec::new();
+    let mut start = 0;
+    let mut depth = 0_usize;
+    let mut quote = None;
+    let mut chars = input.char_indices().peekable();
+    while let Some((position, ch)) = chars.next() {
+        if let Some(delimiter) = quote {
+            if ch == '\\' {
+                chars.next()?;
+            } else if ch == delimiter {
+                if chars.peek().is_some_and(|(_, next)| *next == delimiter) {
+                    chars.next();
+                } else {
+                    quote = None;
+                }
+            }
+            continue;
+        }
+        match ch {
+            '\'' | '`' | '"' => quote = Some(ch),
+            '(' => depth = depth.checked_add(1)?,
+            ')' => depth = depth.checked_sub(1)?,
+            ',' if depth == 0 => {
+                result.push(input[start..position].trim());
+                start = position + 1;
+            }
+            _ => {}
+        }
+    }
+    if depth != 0 || quote.is_some() {
+        return None;
+    }
+    if !input.is_empty() {
+        result.push(input[start..].trim());
+    }
+    (!result.iter().any(|argument| argument.is_empty())).then_some(result)
+}
+
+/// Anonymous native Tuple fields have explicit one-based names in the shared
+/// Struct schema. Named fields must match their native names exactly. Other
+/// Arrow names are never interpreted as an anonymous Tuple.
+pub(super) fn tuple_field_types<'a>(native: &'a str, fields: &[Column]) -> Option<Vec<&'a str>> {
+    let inner = native.strip_prefix("Tuple(")?.strip_suffix(')')?;
+    let args = arguments(inner)?;
+    if args.len() != fields.len() {
+        return None;
+    }
+    let mut types = Vec::with_capacity(fields.len());
+    for (index, (argument, field)) in args.into_iter().zip(fields).enumerate() {
+        if field.table.is_some() {
+            return None;
+        }
+        if field.name == (index + 1).to_string()
+            && super::clickhouse_type_matches(Some(argument), &field.dtype, field.nullable)
+        {
+            types.push(argument);
+            continue;
+        }
+        // Initial named transport accepts ordinary identifiers. Quoted names
+        // remain unsupported until a native identifier-decoding contract exists.
+        let (name, native_type) = argument.split_once(char::is_whitespace)?;
+        if name != field.name
+            || name.is_empty()
+            || !name
+                .chars()
+                .all(|ch| ch.is_ascii_alphanumeric() || ch == '_')
+            || !super::clickhouse_type_matches(
+                Some(native_type.trim()),
+                &field.dtype,
+                field.nullable,
+            )
+        {
+            return None;
+        }
+        types.push(native_type.trim());
+    }
+    Some(types)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use planner_types::pre_asap::DataType;
+
+    #[test]
+    fn tuple_metadata_preserves_names_order_and_nested_types() {
+        let fields = vec![
+            Column::new("ts", DataType::Int64, false),
+            Column::new(
+                "samples",
+                DataType::List {
+                    element: Box::new(Column::new("item", DataType::Float64, true)),
+                },
+                false,
+            ),
+        ];
+        assert_eq!(
+            tuple_field_types("Tuple(ts Int64, samples Array(Nullable(Float64)))", &fields),
+            Some(vec!["Int64", "Array(Nullable(Float64))"])
+        );
+        assert!(
+            tuple_field_types("Tuple(samples Int64, ts Array(Nullable(Float64)))", &fields)
+                .is_none()
+        );
+        assert!(tuple_field_types("Tuple(Int64, Array(Nullable(Float64)))", &fields).is_none());
+        let anonymous = vec![
+            Column::new("1", DataType::Int64, false),
+            Column::new("2", DataType::Utf8, false),
+        ];
+        assert!(tuple_field_types("Tuple(Int64, String)", &anonymous).is_some());
+        assert!(arguments("Map(String, Tuple(Int64, String)), DateTime64(3, 'UTC')").is_some());
+        assert!(arguments("Map(String, Tuple(Int64, String)").is_none());
+    }
+}