feat(query-engine): answer RQE expression trees on POST /api/v1/rqe_query

asap-query-engine/src/drivers/query/servers/http.rs

@@ -14,8 +14,15 @@ use std::time::Instant;
 use tokio::net::TcpListener;
 use tracing::{debug, info, warn};
 
-use crate::drivers::query::adapters::{create_http_adapter, AdapterConfig, HttpProtocolAdapter};
-use crate::engines::{QueryExecutionError, SimpleEngine};
+use crate::drivers::query::adapters::{
+    create_http_adapter, AdapterConfig, HttpProtocolAdapter, PrometheusResponse,
+};
+use crate::engines::{QueryExecutionError, RqeQueryError, SimpleEngine};
+use crate::utils::http::convert_query_result_to_prometheus;
+use asap_types::rqe_expr::RqeExpr;
+
+/// Answers an RQE expression tree, whatever the adapter's query language.
+pub const RQE_QUERY_ENDPOINT: &str = "/api/v1/rqe_query";
 use crate::query_tracker::QueryTracker;
 use crate::stores::Store;
 
@@ -106,6 +113,7 @@ impl HttpServer {
             .route(range_query_endpoint, post(handle_range_query_post))
             .route(runtime_info_path, get(handle_runtime_info))
             .route(runtime_info_path, post(handle_runtime_info))
+            .route(RQE_QUERY_ENDPOINT, post(handle_rqe_query))
             .route("/metrics", get(handle_metrics))
             .with_state(app_state);
 
@@ -143,6 +151,7 @@ impl HttpServer {
             .route(range_query_endpoint, get(handle_range_query))
             .route(range_query_endpoint, post(handle_range_query_post))
             .route(runtime_info_path, get(handle_runtime_info))
+            .route(RQE_QUERY_ENDPOINT, post(handle_rqe_query))
             .with_state(app_state);
 
         let listener = TcpListener::bind("127.0.0.1:0").await?;
@@ -445,6 +454,75 @@ async fn handle_instant_query_post(
     result
 }
 
+/// Body of an RQE query: the tree and its evaluation time in unix seconds.
+#[derive(serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct RqeQueryRequest {
+    expr: RqeExpr,
+    time: f64,
+}
+
+/// A Prometheus-format error response with an explicit HTTP status.
+fn prometheus_error(status: StatusCode, error_type: &str, error: &str) -> Response {
+    (status, Json(PrometheusResponse::error(error_type, error))).into_response()
+}
+
+/// `POST /api/v1/rqe_query`: answers with Prometheus instant-vector JSON. Never
+/// forwards to a fallback backend: an invalid request is a 400, a failed
+/// execution a 500, and a window without data yet an empty result. With
+/// request handling disabled it answers 500 `internal`, as the instant route
+/// does when it has no fallback either.
+async fn handle_rqe_query(State(state): State<AppState>, body: Bytes) -> Response {
+    if !state.config.handle_http_requests {
+        debug!("HTTP request handling is disabled; RQE queries have no fallback");
+        return prometheus_error(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "internal",
+            "RQE query handling is disabled",
+        );
+    }
+    let start_time = Instant::now();
+    let request: RqeQueryRequest = match serde_json::from_slice(&body) {
+        Ok(request) => request,
+        Err(error) => {
+            return prometheus_error(
+                StatusCode::BAD_REQUEST,
+                "bad_data",
+                &format!("invalid RQE query body: {error}"),
+            );
+        }
+    };
+    match state
+        .query_engine
+        .handle_rqe_query(&request.expr, request.time)
+    {
+        Ok((labels, result)) => {
+            info!(
+                "rqe_query time={} destination=asap total_latency_ms={:.2}",
+                request.time,
+                start_time.elapsed().as_secs_f64() * 1000.0
+            );
+            match convert_query_result_to_prometheus(&result, &labels) {
+                Ok(data) => Json(PrometheusResponse::success(data)).into_response(),
+                Err(error) => {
+                    prometheus_error(StatusCode::INTERNAL_SERVER_ERROR, "internal", error)
+                }
+            }
+        }
+        Err(RqeQueryError::BadRequest(error)) => {
+            prometheus_error(StatusCode::BAD_REQUEST, "bad_data", &error)
+        }
+        Err(RqeQueryError::Execution(error)) => {
+            warn!(error = %error, "RQE query execution failed");
+            prometheus_error(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "internal",
+                &error.to_string(),
+            )
+        }
+    }
+}
+
 async fn handle_runtime_info(
     State(state): State<AppState>,
     headers: axum::http::HeaderMap,

asap-query-engine/src/engines/mod.rs

@@ -9,5 +9,5 @@ pub mod window_merger;
 pub use query_result::{InstantVector, QueryResult, RangeVector, RangeVectorElement, Sample};
 #[cfg(feature = "native_query_legacy_test_support")]
 pub use simple_engine::NativeRangeExecutionMode;
-pub use simple_engine::{QueryExecutionError, SimpleEngine};
+pub use simple_engine::{QueryExecutionError, RqeQueryError, SimpleEngine};
 pub use window_merger::{create_window_merger, NaiveMerger, WindowMerger};

asap-query-engine/src/engines/simple_engine/elastic.rs

@@ -44,7 +44,7 @@ impl SimpleEngine {
         // TODO: Figure out how to handle query configuration for ElasticSearch queries.
         let query_config = self.find_query_config(&query)?;
         let agg_info = self
-            .get_aggregation_id_info(&query_config)
+            .get_aggregation_id_info(&query_config.aggregations)
             .map_err(|e| {
                 warn!("{}", e);
                 e

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -1,12 +1,14 @@
 mod elastic;
 mod promql;
+mod rqe;
 mod sql;
 
 use promql::METRIC_NAME_LABEL;
+pub use rqe::RqeQueryError;
 
 use crate::data_model::{
-    AggregationIdInfo, InferenceConfig, KeyByLabelValues, QueryBounds, QueryConfig, QueryLanguage,
-    StreamingConfig,
+    AggregationIdInfo, AggregationReference, InferenceConfig, KeyByLabelValues, QueryBounds,
+    QueryConfig, QueryLanguage, StreamingConfig,
 };
 use crate::engines::query_plan::{
     NodeId, PlanOptions, QueryPlan, QueryPlanExecutionError, QueryPlanNode, QueryPlanRuntime,
@@ -691,6 +693,47 @@ impl SimpleEngine {
         ))
     }
 
+    /// Builds a `QueryExecutionContext` from already-resolved plain values:
+    /// plans the store reads and derives the grouping/aggregated labels from
+    /// the aggregation configs. Shared by the SQL and RQE context builders.
+    #[allow(clippy::too_many_arguments)]
+    fn build_execution_context_tail(
+        &self,
+        metric: &str,
+        timestamps: &QueryTimestamps,
+        metadata: QueryMetadata,
+        agg_info: AggregationIdInfo,
+        spatial_filter: String,
+        query_time: u64,
+    ) -> Result<QueryExecutionContext, String> {
+        let (query_plan, do_merge, value_window_type) =
+            self.create_store_query_plan(metric, timestamps, &agg_info)?;
+
+        let sc = self.streaming_config.read().unwrap().clone();
+        let grouping_labels = sc
+            .get_aggregation_config(agg_info.aggregation_id_for_value)
+            .map(|config| config.grouping_labels.clone())
+            .unwrap_or_else(|| metadata.query_output_labels.clone());
+
+        let aggregated_labels = sc
+            .get_aggregation_config(agg_info.aggregation_id_for_key)
+            .map(|config| config.aggregated_labels.clone())
+            .unwrap_or_else(KeyByLabelNames::empty);
+
+        Ok(QueryExecutionContext {
+            metric: metric.to_string(),
+            metadata,
+            store_plan: query_plan,
+            agg_info,
+            value_window_type,
+            do_merge,
+            spatial_filter,
+            query_time,
+            grouping_labels,
+            aggregated_labels,
+        })
+    }
+
     /// The bucket-map grid width for scanning an aggregation's stored
     /// buckets: `slide_interval_ms`, not `window_size_ms`.
     /// `precompute_engine/window_manager.rs` persists buckets on the
@@ -1771,12 +1814,12 @@ impl SimpleEngine {
         stats.into_iter().next()
     }
 
+    /// Resolves the key and value aggregations a query reads. With two
+    /// aggregations, the set/delta-set one supplies keys and the other values.
     fn get_aggregation_id_info(
         &self,
-        query_config: &QueryConfig,
+        query_config_aggregations: &[AggregationReference],
     ) -> Result<AggregationIdInfo, String> {
-        let query_config_aggregations = &query_config.aggregations;
-
         if query_config_aggregations.is_empty() {
             return Err("Query config has no aggregations defined".to_string());
         }

asap-query-engine/src/engines/simple_engine/promql.rs

@@ -51,7 +51,7 @@ fn detect_scalar_arm<'a>(
 /// two arms don't share the same label set — mirroring DataFusion's
 /// `build_binary_vector_plan`, which fails to resolve a join column that
 /// only exists on one side.
-fn combine_vector_vector(
+pub(super) fn combine_vector_vector(
     lhs_results: Vec<InstantVectorElement>,
     lhs_labels: &[String],
     rhs_results: Vec<InstantVectorElement>,
@@ -83,7 +83,7 @@ fn combine_vector_vector(
 /// Scalar combiner for one binary-expr level (instant query): applies
 /// `op(scalar, value)` or `op(value, scalar)` per `scalar_on_left` to every
 /// element of the vector arm's results.
-fn combine_scalar(
+pub(super) fn combine_scalar(
     vector_results: Vec<InstantVectorElement>,
     scalar: f64,
     op: &promql_parser::parser::token::TokenType,
@@ -119,14 +119,14 @@ impl SimpleEngine {
     /// boundary, unconditionally — mirroring SQL's `align_end_timestamp_sql`.
     /// A no-op in the common case (already-aligned timestamps), a safety net
     /// otherwise, for every PromQL query shape, not just a subset.
-    fn align_end_timestamp_promql(&self, end_timestamp: u64) -> u64 {
+    pub(super) fn align_end_timestamp_promql(&self, end_timestamp: u64) -> u64 {
         let interval_ms = self.data_ingestion_interval_ms;
         if end_timestamp.is_multiple_of(interval_ms) {
             return end_timestamp;
         }
         let aligned = (end_timestamp / interval_ms) * interval_ms;
         warn!(
-            "PromQL query end timestamp {} is not aligned with data ingestion interval of {} ms; \
+            "Query end timestamp {} is not aligned with data ingestion interval of {} ms; \
              aligning down to {}.",
             end_timestamp, interval_ms, aligned
         );
@@ -286,7 +286,7 @@ impl SimpleEngine {
         let match_result = self.find_matching_controller_pattern(arm_ast, planned_subquery)?;
 
         let agg_info = self
-            .get_aggregation_id_info(query_config)
+            .get_aggregation_id_info(&query_config.aggregations)
             .map_err(|e| {
                 warn!("{}", e);
                 e
@@ -1323,7 +1323,7 @@ impl SimpleEngine {
 
         // Resolve aggregation: try pre-configured query_configs first, fall back to capability matching.
         let agg_info: AggregationIdInfo = if let Some(config) = self.find_query_config(query) {
-            self.get_aggregation_id_info(&config)
+            self.get_aggregation_id_info(&config.aggregations)
                 .map_err(|e| {
                     warn!("{}", e);
                     e