diff --git a/Cargo.lock b/Cargo.lock index 964d242a..2c52490b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1198,6 +1198,7 @@ dependencies = [ "base64 0.21.7", "bincode", "chrono", + "chrono-tz", "clap 4.6.1", "control_plane", "crc32fast", diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index 4a9741ab..098cffd4 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -278,6 +278,46 @@ pub struct InstantExecution { } impl QueryPlanEntry { + /// Replace an explicit planner fallback cut with a typed external-exact leaf. + /// The control plane chooses the cut; serving only executes the published DAG. + pub fn bind_external_exact_leaf( + &mut self, + node_id: QueryNodeId, + request: ExternalExactRequest, + ) -> Result<(), QueryPlanError> { + if request.language != self.language { + return Err(QueryPlanError::Invalid( + "external exact language differs from its query plan".into(), + )); + } + if !request.input_contracts.is_empty() { + return Err(QueryPlanError::Invalid( + "leaf binding cannot declare DAG input contracts".into(), + )); + } + match self.nodes.get(&node_id) { + Some(QueryPlanNode::ExactFallback { .. }) => {} + Some(_) => { + return Err(QueryPlanError::Invalid( + "external exact binding must replace a planner fallback cut".into(), + )) + } + None => { + return Err(QueryPlanError::Invalid( + "external exact cut node is absent".into(), + )) + } + } + self.nodes.insert( + node_id, + QueryPlanNode::ExternalExact { + request, + inputs: Vec::new(), + }, + ); + Ok(()) + } + /// Materializations this executable DAG reads, in stable node order. /// Serving uses this set for readiness accounting; it never performs a /// catalog candidate search to reconstruct dependencies. @@ -629,6 +669,14 @@ pub struct ExternalExactRequest { pub language: QueryLanguage, pub expression: String, pub output: ExternalExactOutput, + /// Engine parameters forwarded without embedding transport details in the DAG. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub parameters: BTreeMap, + /// Optional parameter names populated from the query entry's evaluation range. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub start_parameter: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub end_parameter: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] pub input_contracts: Vec, } @@ -1128,6 +1176,9 @@ where language: QueryLanguage::PromQl, expression: aggregate.expr.to_string(), output: ExternalExactOutput::InstantVector, + parameters: BTreeMap::new(), + start_parameter: None, + end_parameter: None, input_contracts: vec![ExternalExactInput::CandidateMembership { item_label, }], diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index cd7a7a4e..5df1d0c9 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -51,6 +51,7 @@ tracing.workspace = true tracing-subscriber.workspace = true clap.workspace = true chrono.workspace = true +chrono-tz = "0.10" promql-parser.workspace = true tokio.workspace = true arc-swap.workspace = true 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 fea8e52d..fbc43cbb 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 @@ -5,12 +5,15 @@ use axum::{ body::Bytes, http::{HeaderMap, HeaderValue, StatusCode}, }; -use std::sync::Arc; +use std::{collections::BTreeMap, sync::Arc}; use super::{ clickhouse_result_adapter::ClickHouseFormat, - execution::{execute_sql_dag, ClickHouseDagFallback, ClickHouseDagOutcome}, - fallback::ClickHouseRawResponse, + execution::{ + execute_sql_dag_with_external, ClickHouseDagFallback, ClickHouseDagOutcome, + PreparedExternalLeaves, + }, + fallback::{ClickHouseExactBackend, ClickHouseRawResponse}, request::ClickHouseQueryRequest, server::{ ClickHouseAccelerationFallback, ClickHouseAccelerationOutcome, ClickHouseAccelerator, @@ -21,6 +24,7 @@ use crate::storage_engines::sketch_db::index::SketchStore; pub struct CatalogClickHouseAccelerator { pub store: Arc, active_physical_plan: Option, + exact_backend: Option>, } impl CatalogClickHouseAccelerator { @@ -28,6 +32,7 @@ impl CatalogClickHouseAccelerator { Self { store, active_physical_plan: None, + exact_backend: None, } } @@ -39,6 +44,82 @@ impl CatalogClickHouseAccelerator { accelerator.active_physical_plan = Some(active); accelerator } + + pub fn with_exact_backend(mut self, exact_backend: Arc) -> Self { + self.exact_backend = Some(exact_backend); + self + } + + async fn prepare_external_exact( + &self, + entry: &control_plane::query_plan::QueryPlanEntry, + start_ms: u64, + end_ms: u64, + request_context: &ClickHouseQueryRequest, + ) -> Result { + let mut prepared = PreparedExternalLeaves::new(); + let leaves = entry.nodes.iter().filter_map(|(id, node)| match node { + control_plane::query_plan::QueryPlanNode::ExternalExact { request, inputs } + if request.language == asap_types::QueryLanguage::ClickHouseSql + && inputs.is_empty() => + { + Some((*id, request)) + } + _ => None, + }); + for (id, bound) in leaves { + let backend = self + .exact_backend + .as_ref() + .ok_or_else(|| "ClickHouse exact subtree endpoint unavailable".to_owned())?; + let schema = match &bound.output { + control_plane::query_plan::ExternalExactOutput::Relation { schema } => { + serde_json::from_value(schema.clone()).map_err(|error| error.to_string())? + } + _ => return Err("ClickHouse exact subtree must produce a relation".into()), + }; + let mut parameters = bound + .parameters + .iter() + .map(|(name, value)| (format!("param_{name}"), value.clone())) + .collect::>(); + if let Some(name) = &bound.start_parameter { + parameters.insert(format!("param_{name}"), start_ms.to_string()); + } + if let Some(name) = &bound.end_parameter { + parameters.insert(format!("param_{name}"), end_ms.to_string()); + } + parameters.insert("default_format".into(), "JSONCompact".into()); + if let Some(database) = request_context.database() { + parameters.insert("database".into(), database.into()); + } + let request = ClickHouseQueryRequest { + method: axum::http::Method::POST, + sql: bound.expression.clone(), + body: Bytes::from(bound.expression.clone()), + parameters, + headers: request_context.headers.clone(), + }; + let response = backend + .execute(&request) + .await + .map_err(|error| error.to_string())?; + if !response.status.is_success() { + return Err(format!( + "ClickHouse external subtree returned HTTP {}", + response.status + )); + } + let mut relation = super::relational_adapter::ClickHouseRelation::from_json_compact( + &schema, + &response.body, + ) + .map_err(|error| error.to_string())?; + relation.coverage = Some((start_ms, end_ms)); + prepared.insert(id, relation); + } + Ok(prepared) + } } fn requested_format(request: &ClickHouseQueryRequest) -> Result { @@ -109,10 +190,22 @@ impl ClickHouseAccelerator for CatalogClickHouseAccelerator { ), ); }; - match execute_sql_dag( + let prepared = match self + .prepare_external_exact(entry, range.start_ms, range.end_ms, request) + .await + { + Ok(prepared) => prepared, + Err(error) => { + return ClickHouseAccelerationOutcome::Fallback( + ClickHouseAccelerationFallback::Execution(error), + ) + } + }; + match execute_sql_dag_with_external( self.store.as_ref(), entry, catalog.as_ref(), + &prepared, range.start_ms, range.end_ms, range.cumulative, @@ -164,10 +257,36 @@ mod tests { use asap_types::{AggregationType, KeyByLabelNames, PrecomputeMaterialization, WindowKind}; use axum::http::Method; use control_plane::query_plan::{ - ClickHousePlanningContext, ExactReadout, FallbackPolicy, FixedEvaluationRange, - InstantExecution, MaterializationBinding, PhysicalGrouping, QueryLanguage, QueryNodeId, - QueryPlan, QueryPlanEntry, QueryPlanNode, + ClickHousePlanningContext, ExactReadout, ExternalExactOutput, ExternalExactRequest, + FallbackPolicy, FixedEvaluationRange, InstantExecution, MaterializationBinding, + PhysicalGrouping, QueryLanguage, QueryNodeId, QueryPlan, QueryPlanEntry, QueryPlanNode, }; + + struct FixedExactSubtree; + + #[async_trait] + impl ClickHouseExactBackend for FixedExactSubtree { + async fn execute( + &self, + request: &ClickHouseQueryRequest, + ) -> Result + { + assert_eq!(request.parameters.get("param_from"), Some(&"0".into())); + assert_eq!(request.parameters.get("param_to"), Some(&"2000".into())); + Ok(ClickHouseRawResponse { + status: StatusCode::OK, + headers: HeaderMap::new(), + body: Bytes::from_static(br#"{"meta":[{"name":"timestamp","type":"Int64"},{"name":"divisor","type":"Float64"}],"data":[[2000,10.0]],"rows":1}"#), + }) + } + + async fn ping( + &self, + ) -> Result + { + unreachable!() + } + } use planner_types::{ post_asap::{SummaryFamilyType, SummaryField, SummarySchema, ValueOperation}, pre_asap::{ @@ -624,4 +743,115 @@ mod tests { .unwrap(); assert_eq!(accelerated_value, exact_value); } + + #[tokio::test] + async fn summary_and_external_exact_leaf_compose_in_one_query_dag() { + let (accelerator, request) = fixture(2_000).await; + let physical = accelerator + .active_physical_plan + .as_ref() + .unwrap() + .snapshot(); + let mut entry = physical.query_plan.entries.values().next().unwrap().clone(); + let left = QueryNodeId(1); + let external = QueryNodeId(6); + let join = QueryNodeId(7); + let project = QueryNodeId(8); + let left_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("value", DataType::Float64), + ]); + let right_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("divisor", DataType::Float64), + ]); + let joined_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("value", DataType::Float64), + ("timestamp", DataType::Timestamp), + ("divisor", DataType::Float64), + ]); + let output_schema = relation_schema(&[ + ("timestamp", DataType::Timestamp), + ("ratio", DataType::Float64), + ]); + entry + .nodes + .retain(|id, _| *id == QueryNodeId(0) || *id == left); + entry.nodes.insert(external, QueryPlanNode::ExternalExact { + request: ExternalExactRequest { + language: QueryLanguage::ClickHouseSql, + expression: "SELECT toInt64(2000) AS timestamp, toFloat64(10) AS divisor WHERE {from:UInt64} <= {to:UInt64}".into(), + output: ExternalExactOutput::Relation { schema: serde_json::to_value(&right_schema).unwrap() }, + parameters: BTreeMap::new(), + start_parameter: Some("from".into()), + end_parameter: Some("to".into()), + input_contracts: vec![], + }, + inputs: vec![], + }); + entry.nodes.insert( + join, + QueryPlanNode::RelationalJoin { + inputs: [left, external], + join_kind: planner_types::pre_asap::JoinKind::Inner, + pred: serde_json::to_value(Predicate(Rc::new(QueryExpr::Compare { + left: Rc::new(QueryExpr::Column(0)), + op: CompareOpKind::Eq, + right: Rc::new(QueryExpr::Column(2)), + }))) + .unwrap(), + left_schema, + right_schema, + output_schema: joined_schema.clone(), + }, + ); + entry.nodes.insert( + project, + QueryPlanNode::Relational { + input: join, + operation: serde_json::to_value(ValueOperation::Project { + cols: vec![ + ProjectItem { + alias: Some("timestamp".into()), + expr: QueryExpr::Column(0), + }, + ProjectItem { + alias: Some("ratio".into()), + expr: QueryExpr::Arithmetic { + op: ArithmeticOpKind::Div, + left: Rc::new(QueryExpr::Column(1)), + right: Rc::new(QueryExpr::Column(3)), + }, + }, + ], + qualifier: None, + }) + .unwrap(), + input_schema: joined_schema, + output_schema, + }, + ); + entry.root = project; + let accelerator = accelerator.with_exact_backend(Arc::new(FixedExactSubtree)); + let prepared = accelerator + .prepare_external_exact(&entry, 0, 2_000, &request) + .await + .unwrap(); + let ClickHouseDagOutcome::Accelerated(result) = execute_sql_dag_with_external( + accelerator.store.as_ref(), + &entry, + physical.summary_catalog.as_ref().unwrap(), + &prepared, + 0, + 2_000, + true, + ) else { + panic!("mixed summary/external DAG should execute") + }; + assert_eq!( + String::from_utf8(result.encode(ClickHouseFormat::TabSeparated).unwrap()).unwrap(), + "1970-01-01T00:00:02\t0.5\n" + ); + } } diff --git a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs index c4424e2b..8215af48 100644 --- a/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs +++ b/data_plane/src/query_engines/asap_clickhouse_query_engine/execution.rs @@ -12,7 +12,9 @@ use crate::{ use asap_types::summary_catalog::SummaryCatalog; use control_plane::query_plan::{QueryNodeId, QueryPlanEntry, QueryPlanNode}; use planner_types::post_asap::ValueOperation; -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; + +pub type PreparedExternalLeaves = BTreeMap; #[derive(Debug, Clone, PartialEq, Eq)] pub enum ClickHouseDagFallback { @@ -53,11 +55,31 @@ fn execute_relation_subtree( entry: &QueryPlanEntry, root: QueryNodeId, expected_schema: &planner_types::post_asap::SummarySchema, + prepared: &PreparedExternalLeaves, t0_ms: u64, t1_ms: u64, is_cumulative: bool, ) -> Result { match entry.nodes.get(&root) { + Some(QueryPlanNode::ExternalExact { request, .. }) => { + if request.language != asap_types::QueryLanguage::ClickHouseSql { + return Err("ClickHouse DAG contains an external leaf for another language".into()); + } + let control_plane::query_plan::ExternalExactOutput::Relation { schema } = + &request.output + else { + return Err("ClickHouse external leaf must declare relation output".into()); + }; + let declared: planner_types::post_asap::SummarySchema = + serde_json::from_value(schema.clone()).map_err(|error| error.to_string())?; + if &declared != expected_schema { + return Err("external exact leaf schema differs from its parent edge".into()); + } + prepared + .get(&root) + .cloned() + .ok_or_else(|| "published external exact leaf was not prepared".into()) + } Some(QueryPlanNode::Relational { input, operation, @@ -69,6 +91,7 @@ fn execute_relation_subtree( entry, *input, input_schema, + prepared, t0_ms, t1_ms, is_cumulative, @@ -91,6 +114,7 @@ fn execute_relation_subtree( entry, inputs[0], left_schema, + prepared, t0_ms, t1_ms, is_cumulative, @@ -100,6 +124,7 @@ fn execute_relation_subtree( entry, inputs[1], right_schema, + prepared, t0_ms, t1_ms, is_cumulative, @@ -185,6 +210,26 @@ pub fn execute_sql_dag( t0_ms: u64, t1_ms: u64, is_cumulative: bool, +) -> ClickHouseDagOutcome { + execute_sql_dag_with_external( + index, + entry, + sds, + &PreparedExternalLeaves::new(), + t0_ms, + t1_ms, + is_cumulative, + ) +} + +pub fn execute_sql_dag_with_external( + index: &SketchStore, + entry: &QueryPlanEntry, + sds: &SummaryCatalog, + prepared: &PreparedExternalLeaves, + t0_ms: u64, + t1_ms: u64, + is_cumulative: bool, ) -> ClickHouseDagOutcome { if let Err(error) = validate_payload(Some(sds), entry, sds.plan_id, sds.plan_version) { return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( @@ -195,14 +240,31 @@ pub fn execute_sql_dag( ids.iter().any(|id| { matches!( entry.nodes.get(id), - Some(QueryPlanNode::RelationalJoin { .. }) + Some(QueryPlanNode::RelationalJoin { .. } | QueryPlanNode::ExternalExact { .. }) ) }) }); if has_relational_join { let root_schema = match entry.nodes.get(&entry.root) { Some(QueryPlanNode::Relational { output_schema, .. }) - | Some(QueryPlanNode::RelationalJoin { output_schema, .. }) => output_schema, + | Some(QueryPlanNode::RelationalJoin { output_schema, .. }) => output_schema.clone(), + Some(QueryPlanNode::ExternalExact { request, .. }) => { + let control_plane::query_plan::ExternalExactOutput::Relation { schema } = + &request.output + else { + return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( + "ClickHouse external root must declare relation output".into(), + )); + }; + match serde_json::from_value(schema.clone()) { + Ok(schema) => schema, + Err(error) => { + return ClickHouseDagOutcome::Fallback( + ClickHouseDagFallback::UnsupportedPlan(error.to_string()), + ) + } + } + } _ => { return ClickHouseDagOutcome::Fallback(ClickHouseDagFallback::UnsupportedPlan( "relational join plan root has no relation schema".into(), @@ -213,7 +275,8 @@ pub fn execute_sql_dag( index, entry, entry.root, - root_schema, + &root_schema, + prepared, t0_ms, t1_ms, is_cumulative, 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 7ddef54c..d33799a7 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 @@ -9,6 +9,7 @@ use arrow::{ datatypes::{DataType as ArrowDataType, Field, Schema}, record_batch::RecordBatch, }; +use chrono::{DateTime, NaiveDateTime, TimeZone}; use planner_types::{ post_asap::{SummaryFamilyType, SummaryNode, SummarySchema, ValueOperation}, pre_asap::{ArithmeticOpKind, CompareOpKind, DataType, QueryExpr, ScalarValue, SortKey}, @@ -46,7 +47,92 @@ enum Cell { Timestamp(i64), } -#[derive(Debug)] +fn json_cell( + value: &serde_json::Value, + dtype: &DataType, + nullable: bool, + clickhouse_type: &str, +) -> Result { + if value.is_null() && nullable { + return Ok(Cell::Null); + } + let invalid = || { + ClickHouseRelationalError::Invalid(format!( + "external value {value} does not match {dtype:?}" + )) + }; + match dtype { + 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 + .as_str() + .map(|value| Cell::Utf8(value.into())) + .ok_or_else(invalid), + DataType::Bool => value.as_bool().map(Cell::Bool).ok_or_else(invalid), + DataType::Timestamp => parse_clickhouse_timestamp(value, clickhouse_type) + .map(Cell::Timestamp) + .ok_or_else(invalid), + } +} + +fn parse_clickhouse_timestamp(value: &serde_json::Value, clickhouse_type: &str) -> Option { + let clickhouse_type = clickhouse_type + .strip_prefix("Nullable(") + .and_then(|value| value.strip_suffix(')')) + .unwrap_or(clickhouse_type); + if clickhouse_type == "Int64" { + return value.as_i64(); + } + let text = value.as_str()?; + if let Ok(timestamp) = DateTime::parse_from_rfc3339(text) { + return Some(timestamp.timestamp_millis()); + } + let (scale, timezone) = if clickhouse_type == "DateTime" { + (0, "UTC") + } else if let Some(timezone) = clickhouse_type + .strip_prefix("DateTime(") + .and_then(|value| value.strip_suffix(')')) + { + (0, timezone.trim().trim_matches('\'')) + } else { + let args = clickhouse_type + .strip_prefix("DateTime64(")? + .strip_suffix(')')?; + let mut args = args.split(',').map(str::trim); + let scale = args.next()?.parse::().ok()?; + if scale > 9 { + return None; + } + let timezone = args.next().unwrap_or("UTC").trim_matches('\''); + if args.next().is_some() { + return None; + } + (scale, timezone) + }; + let fraction_digits = text + .split_once('.') + .map(|(_, fraction)| fraction.len()) + .unwrap_or(0); + if fraction_digits != scale { + return None; + } + let naive = NaiveDateTime::parse_from_str( + text, + if scale == 0 { + "%Y-%m-%d %H:%M:%S" + } else { + "%Y-%m-%d %H:%M:%S%.f" + }, + ) + .ok()?; + let timezone: chrono_tz::Tz = timezone.parse().ok()?; + timezone + .from_local_datetime(&naive) + .single() + .map(|value| value.timestamp_millis()) +} + +#[derive(Clone, Debug)] pub struct ClickHouseRelation { rows: Vec>, fields: Vec<(String, DataType, bool)>, @@ -54,6 +140,81 @@ pub struct ClickHouseRelation { } impl ClickHouseRelation { + pub fn from_json_compact( + schema: &SummarySchema, + body: &[u8], + ) -> Result { + let document: serde_json::Value = serde_json::from_slice(body) + .map_err(|error| ClickHouseRelationalError::Invalid(error.to_string()))?; + let meta = document + .get("meta") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| { + ClickHouseRelationalError::Invalid( + "ClickHouse JSONCompact response has no typed metadata".into(), + ) + })?; + let data = document + .get("data") + .and_then(serde_json::Value::as_array) + .ok_or_else(|| { + ClickHouseRelationalError::Invalid( + "ClickHouse JSONCompact response has no data rows".into(), + ) + })?; + let fields = fields_from_schema(schema); + if meta.len() != fields.len() + || meta + .iter() + .zip(&fields) + .any(|(actual, (name, dtype, nullable))| { + actual.get("name").and_then(serde_json::Value::as_str) != Some(name) + || !clickhouse_type_matches( + actual.get("type").and_then(serde_json::Value::as_str), + dtype, + *nullable, + ) + }) + { + return Err(ClickHouseRelationalError::Invalid( + "ClickHouse external metadata differs from its planned schema".into(), + )); + } + let mut rows = Vec::with_capacity(data.len()); + for encoded in data { + let encoded = encoded.as_array().ok_or_else(|| { + ClickHouseRelationalError::Invalid( + "ClickHouse JSONCompact row is not an array".into(), + ) + })?; + if encoded.len() != fields.len() { + return Err(ClickHouseRelationalError::Invalid( + "ClickHouse external row differs from its planned schema".into(), + )); + } + rows.push( + encoded + .iter() + .zip(&fields) + .zip(meta) + .map(|((value, (_, dtype, nullable)), metadata)| { + json_cell( + value, + dtype, + *nullable, + metadata["type"].as_str().expect("metadata validated"), + ) + }) + .collect::, _>>()?, + ); + } + Ok(Self { + rows, + fields, + coverage: None, + }) + } + pub fn from_series_rows( schema: &SummarySchema, series: Vec<(BTreeMap, Vec<(i64, f64)>)>, @@ -90,6 +251,35 @@ impl ClickHouseRelation { } } +fn clickhouse_type_matches(actual: Option<&str>, expected: &DataType, nullable: bool) -> bool { + let Some(mut actual) = actual else { + return false; + }; + if nullable { + let Some(inner) = actual + .strip_prefix("Nullable(") + .and_then(|value| value.strip_suffix(')')) + else { + return false; + }; + actual = inner; + } else if actual.starts_with("Nullable(") { + return false; + } + match expected { + DataType::Int64 => actual == "Int64", + DataType::Float64 => actual == "Float64", + DataType::Utf8 => actual == "String", + DataType::Bool => actual == "Bool", + DataType::Timestamp => { + actual == "Int64" + || actual == "DateTime" + || actual.starts_with("DateTime(") + || actual.starts_with("DateTime64(") + } + } +} + pub struct ClickHouseRelationalAdapter; impl ClickHouseRelationalAdapter { diff --git a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs index 7de574de..ab6ebc68 100644 --- a/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs +++ b/data_plane/src/query_engines/asap_query_engine/exact_subqueries.rs @@ -457,6 +457,9 @@ mod tests { language: QueryLanguage::PromQl, expression: query.into(), output: control_plane::query_plan::ExternalExactOutput::InstantVector, + parameters: BTreeMap::new(), + start_parameter: None, + end_parameter: None, input_contracts: vec![ExternalExactInput::CandidateMembership { item_label: "job".into(), }],