diff --git a/control_plane/src/query_plan/clickhouse_exact.rs b/control_plane/src/query_plan/clickhouse_exact.rs index ea7c7c55..fa227841 100644 --- a/control_plane/src/query_plan/clickhouse_exact.rs +++ b/control_plane/src/query_plan/clickhouse_exact.rs @@ -78,6 +78,7 @@ fn scalar(expr: &QueryExpr, schema: &Schema) -> Result { "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![]); diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs index cdecb879..8ddb70e9 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter.rs +++ b/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::, _>>()? + .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::, _>>()? + .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 diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter/collection.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/relational_adapter/collection.rs new file mode 100644 index 00000000..2bb43e7b --- /dev/null +++ b/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> { + 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> { + 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()); + } +}