diff --git a/Cargo.lock b/Cargo.lock index cd77ff93..c0dfaf2b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,7 +373,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=0402384e589df6e087d6d2d22b463ddc2eea0774#0402384e589df6e087d6d2d22b463ddc2eea0774" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5e9396033a5a5d5f350bfa675d12770c972ee991#5e9396033a5a5d5f350bfa675d12770c972ee991" dependencies = [ "asap-types", "serde", @@ -384,7 +384,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=0402384e589df6e087d6d2d22b463ddc2eea0774#0402384e589df6e087d6d2d22b463ddc2eea0774" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5e9396033a5a5d5f350bfa675d12770c972ee991#5e9396033a5a5d5f350bfa675d12770c972ee991" dependencies = [ "asap-types", "promql-parser", @@ -393,7 +393,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=0402384e589df6e087d6d2d22b463ddc2eea0774#0402384e589df6e087d6d2d22b463ddc2eea0774" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5e9396033a5a5d5f350bfa675d12770c972ee991#5e9396033a5a5d5f350bfa675d12770c972ee991" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -415,12 +415,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=0402384e589df6e087d6d2d22b463ddc2eea0774#0402384e589df6e087d6d2d22b463ddc2eea0774" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5e9396033a5a5d5f350bfa675d12770c972ee991#5e9396033a5a5d5f350bfa675d12770c972ee991" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=0402384e589df6e087d6d2d22b463ddc2eea0774#0402384e589df6e087d6d2d22b463ddc2eea0774" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=5e9396033a5a5d5f350bfa675d12770c972ee991#5e9396033a5a5d5f350bfa675d12770c972ee991" dependencies = [ "serde", "serde_json", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 7cb29606..26453434 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -76,8 +76,8 @@ asap_types.workspace = true # scaffolding, unaware that `data_plane`'s `summary_executor.rs` in *this* # repo is a real one. Vendored locally instead of chased upstream -- see # `data_plane/src/query_engines/asap_query_engine/summary_exec.rs`. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } # L1 adoption (design-target-architecture.md Part B): the PromQL front # end itself, replacing control_plane's own query_parser/promql.rs. @@ -85,8 +85,8 @@ asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = # `planner-types`/`asap-aware-mapping` above -- these three MUST move # together (two revs of the same upstream repo's types in one workspace # resolve to distinct Rust types that won't unify). -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } -asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } +asap-frontend-sql = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/query_plan/clickhouse_exact.rs b/control_plane/src/query_plan/clickhouse_exact.rs index 85f47609..ea7c7c55 100644 --- a/control_plane/src/query_plan/clickhouse_exact.rs +++ b/control_plane/src/query_plan/clickhouse_exact.rs @@ -77,7 +77,7 @@ fn scalar(expr: &QueryExpr, schema: &Schema) -> Result { let function = match name.as_str() { "map" => "map", "mapconcat" => "mapConcat", - "asap_map_access" => "arrayElement", + "asap_map_access" | "asap_element_access" => "arrayElement", _ => return Err(format!("unsupported exact scalar function {name}")), }; expr.scalar_type(schema).map_err(|e| e.to_string())?; @@ -301,6 +301,28 @@ mod tests { assert!(!sql.contains("{from:")); assert!(!sql.contains("{to:")); } + #[test] + fn typed_list_access_renders_native_element_lookup() { + let schema = Schema::new(vec![Column::new( + "samples", + DataType::List { + element: Box::new(Column::new("item", DataType::Float64, false)), + }, + false, + )]); + let expr = QueryExpr::FunctionCall { + name: "asap_element_access".into(), + args: vec![ + QueryExpr::Column(0), + QueryExpr::Literal(ScalarValue::Int64(-1)), + ], + }; + assert_eq!( + scalar(&expr, &schema).unwrap(), + "arrayElement(`samples`, -1)" + ); + } + #[test] fn unsupported_scalar_is_not_forwarded_as_arbitrary_native_code() { let schema = Schema::new(vec![]); diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index 671d34ca..d6480e60 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -34,4 +34,4 @@ sha2 = "0.10" # exactly (`control_plane/Cargo.toml`) -- two different revs of the same # git dependency in one workspace resolve to two distinct Rust types that # won't unify. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 60ae2ef0..cc531049 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -39,8 +39,8 @@ sha2 = "0.10" # reduction: Reduction, .. }`) are `pre_asap` types, in the same crate now # (not a separate `asap-ir` import). Query serving consumes the compiled # QueryPlan; these types are used at physical-plan compilation boundaries. -planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } -asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } +planner-types = { package = "asap-types", git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } +asap-frontend-promql = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } # Shared external (workspace) serde.workspace = true @@ -133,7 +133,7 @@ fs2 = "0.4" # none of them. [dev-dependencies] -asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "0402384e589df6e087d6d2d22b463ddc2eea0774" } +asap-aware-mapping = { git = "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/ProjectASAP/ASAPPlanner", rev = "5e9396033a5a5d5f350bfa675d12770c972ee991" } tempfile = "3.20.0" criterion = { version = "0.5", features = ["html_reports"] } tokio-tungstenite = "0.21" diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs index 907b7fea..8e34ceaf 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/accelerator.rs @@ -111,6 +111,9 @@ impl CatalogClickHouseAccelerator { "0".into(), ); parameters.insert("output_format_json_quote_64bit_integers".into(), "0".into()); + // Preserve the distinction between NULL and unsupported NaN/Inf. + // The typed decoder rejects quoted non-finite values and falls back. + parameters.insert("output_format_json_quote_denormals".into(), "1".into()); if let Some(database) = request_context.database() { parameters.insert("database".into(), database.into()); } @@ -328,6 +331,10 @@ mod tests { request: &ClickHouseQueryRequest, ) -> Result { + assert_eq!( + request.parameters.get("output_format_json_quote_denormals"), + Some(&"1".into()) + ); assert_eq!(request.parameters.get("param_from"), Some(&"0".into())); assert_eq!(request.parameters.get("param_to"), Some(&"2000".into())); Ok(ClickHouseRawResponse { 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 15945be2..cdecb879 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 @@ -41,6 +41,7 @@ enum Cell { Bool(bool), Timestamp(i64), Map(Vec<(Cell, Cell)>), + List(Arc<[Cell]>), } fn json_cell( @@ -60,9 +61,23 @@ fn json_cell( match dtype { DataType::Null if value.is_null() => Ok(Cell::Null), DataType::Null => Err(invalid()), - DataType::List { .. } | DataType::Struct { .. } => Err( - ClickHouseRelationalError::Unsupported("collection value transport".into()), - ), + DataType::List { element } => { + let item_type = clickhouse_type + .strip_prefix("Array(") + .and_then(|inner| inner.strip_suffix(')')) + .ok_or_else(invalid)?; + let items = value.as_array().ok_or_else(invalid)?; + Ok(Cell::List( + items + .iter() + .map(|item| json_cell(item, &element.dtype, element.nullable, item_type)) + .collect::, _>>()? + .into(), + )) + } + DataType::Struct { .. } => Err(ClickHouseRelationalError::Unsupported( + "struct value transport".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 @@ -307,7 +322,16 @@ fn clickhouse_type_matches(actual: Option<&str>, expected: &DataType, nullable: } match expected { DataType::Null => actual == "Nothing", - DataType::List { .. } | DataType::Struct { .. } => false, + DataType::List { element } => { + !nullable + && actual + .strip_prefix("Array(") + .and_then(|inner| inner.strip_suffix(')')) + .is_some_and(|inner| { + clickhouse_type_matches(Some(inner), &element.dtype, element.nullable) + }) + } + DataType::Struct { .. } => false, DataType::Int64 => actual == "Int64", DataType::Float64 => actual == "Float64", DataType::Utf8 => actual == "String", @@ -428,11 +452,17 @@ impl ClickHouseRelationalAdapter { } for row in &input.rows { for key in keys { - if contains_nan(&eval(&key.expr, row, &schema)?) { + let value = eval(&key.expr, row, &schema)?; + if contains_nan(&value) { return Err(ClickHouseRelationalError::Unsupported( "NaN sort key".into(), )); } + if !matches!(value, Cell::Null) && cell_cmp(&value, &value).is_none() { + return Err(ClickHouseRelationalError::Unsupported( + "unsupported sort key value type".into(), + )); + } } } input @@ -548,7 +578,50 @@ fn eval( } QueryExpr::FunctionCall { name, args } => { use planner_types::pre_asap::scalar_signature::MapScalarFunction; - let function = MapScalarFunction::from_name(name).ok_or_else(|| { + if name.eq_ignore_ascii_case("asap_element_access") { + let (output_type, _) = expr + .scalar_type(schema) + .map_err(|error| ClickHouseRelationalError::Invalid(error.to_string()))?; + if let DataType::List { element } = args[0] + .scalar_type(schema) + .map_err(|error| ClickHouseRelationalError::Invalid(error.to_string()))? + .0 + { + let Cell::List(values) = eval(&args[0], row, schema)? else { + return Err(ClickHouseRelationalError::Invalid( + "array access input".into(), + )); + }; + let index = match eval(&args[1], row, schema)? { + Cell::Null => return Ok(Cell::Null), + Cell::Int64(index) => index, + _ => { + return Err(ClickHouseRelationalError::Invalid( + "array access index".into(), + )) + } + }; + let offset = if index > 0 { + usize::try_from(index - 1).ok() + } else if index < 0 { + usize::try_from(index.unsigned_abs()) + .ok() + .and_then(|distance| values.len().checked_sub(distance)) + } else { + None + }; + return match offset.and_then(|offset| values.get(offset)) { + Some(value) => Ok(value.clone()), + None => default_collection_element(&output_type, element.nullable), + }; + } + } + let function = (if name.eq_ignore_ascii_case("asap_element_access") { + Some(MapScalarFunction::Access) + } else { + MapScalarFunction::from_name(name) + }) + .ok_or_else(|| { ClickHouseRelationalError::Unsupported(format!("scalar function {name}")) })?; expr.scalar_type(schema) @@ -619,7 +692,7 @@ fn eval( else { unreachable!() }; - default_map_value(&value, value_nullable) + default_collection_element(&value, value_nullable) } } } @@ -629,7 +702,10 @@ fn eval( } } -fn default_map_value(dtype: &DataType, nullable: bool) -> Result { +fn default_collection_element( + dtype: &DataType, + nullable: bool, +) -> Result { if nullable { return Ok(Cell::Null); } @@ -640,9 +716,10 @@ fn default_map_value(dtype: &DataType, nullable: bool) -> Result Cell::Utf8(String::new()), DataType::Bool => Cell::Bool(false), DataType::Map { .. } => Cell::Map(Vec::new()), + DataType::List { .. } => Cell::List(Arc::from([])), _ => { return Err(ClickHouseRelationalError::Unsupported( - "map missing-key default type".into(), + "collection missing-element default type".into(), )) } }) @@ -768,6 +845,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::Map(entries) => entries .iter() .any(|(key, value)| contains_nan(key) || contains_nan(value)), @@ -974,6 +1052,141 @@ mod tests { }; use std::rc::Rc; + #[test] + fn decodes_declared_array_elements_without_losing_nullability() { + use planner_types::pre_asap::Column; + let dtype = DataType::List { + element: Box::new(Column { + name: "item".into(), + dtype: DataType::Int64, + nullable: true, + table: None, + }), + }; + assert!(clickhouse_type_matches( + Some("Array(Nullable(Int64))"), + &dtype, + false + )); + assert!(!clickhouse_type_matches( + Some("Array(Int64)"), + &dtype, + false + )); + assert!(!clickhouse_type_matches( + Some("Nullable(Array(Nullable(Int64)))"), + &dtype, + true + )); + let value = json_cell( + &serde_json::json!([9007199254740993_i64, null, -7]), + &dtype, + false, + "Array(Nullable(Int64))", + ) + .unwrap(); + let Cell::List(items) = &value else { + panic!("expected list") + }; + assert_eq!( + items.as_ref(), + &[Cell::Int64(9007199254740993), Cell::Null, Cell::Int64(-7)] + ); + let Cell::List(copy) = value.clone() else { + unreachable!() + }; + assert!(Arc::ptr_eq(items, ©)); + assert!(json_cell( + &serde_json::json!(["wrong"]), + &dtype, + false, + "Array(Nullable(Int64))" + ) + .is_err()); + } + + #[test] + fn nonfinite_external_array_values_cannot_become_nulls() { + use planner_types::pre_asap::Column; + let dtype = DataType::List { + element: Box::new(Column::new("item", DataType::Float64, true)), + }; + for value in ["inf", "-inf", "nan"] { + assert!(json_cell( + &serde_json::json!(value), + &DataType::Float64, + true, + "Nullable(Float64)" + ) + .is_err()); + assert!(json_cell( + &serde_json::json!([value]), + &dtype, + false, + "Array(Nullable(Float64))" + ) + .is_err()); + } + assert_eq!( + json_cell( + &serde_json::json!([null]), + &dtype, + false, + "Array(Nullable(Float64))" + ) + .unwrap(), + Cell::List(vec![Cell::Null].into()) + ); + } + + #[test] + fn array_access_uses_signed_indices_and_element_defaults() { + use planner_types::pre_asap::{Column, Schema}; + let function = |name: &str, args| QueryExpr::FunctionCall { + name: name.into(), + args, + }; + let dtype = DataType::List { + element: Box::new(Column::new("item", DataType::Int64, false)), + }; + let schema = Schema::new(vec![ + Column::new("items", dtype, false), + Column::new("index", DataType::Int64, true), + ]); + let access = function( + "asap_element_access", + vec![QueryExpr::Column(0), QueryExpr::Column(1)], + ); + let items = Cell::List(vec![Cell::Int64(10), Cell::Int64(20)].into()); + for (index, expected) in [ + (1, 10), + (2, 20), + (-1, 20), + (-2, 10), + (0, 0), + (3, 0), + (i64::MIN, 0), + (i64::MAX, 0), + ] { + assert_eq!( + eval(&access, &[items.clone(), Cell::Int64(index)], &schema).unwrap(), + Cell::Int64(expected) + ); + } + assert_eq!( + eval(&access, &[items, Cell::Null], &schema).unwrap(), + Cell::Null + ); + let zero = function( + "asap_element_access", + vec![ + QueryExpr::Column(0), + QueryExpr::Literal(ScalarValue::Int64(0)), + ], + ); + assert!(eval(&zero, &[Cell::List(Arc::from([])), Cell::Int64(0)], &schema).is_err()); + } + fn schema(fields: &[(&str, DataType)]) -> SummarySchema { SummarySchema { fields: fields