Normalizer-based SQL pattern recognition (planner + query engine + ingest)

asap-common/dependencies/rs/asap_types/src/computed_label.rs

@@ -0,0 +1,37 @@
+use serde::{Deserialize, Serialize};
+
+/// Config for a computed label: a metadata column derived from another raw
+/// column at ingest time rather than read directly off the row, e.g.
+/// extracting the origin ASN (`select: last`) or every ASN (`select: all`,
+/// via `type: token_explode`) out of a space-separated `as_path` string.
+///
+/// Shared between asap-planner-rs (which detects a nested-subquery SQL shape
+/// like `arrayFilter(...) AS as_path_array ... as_path_array[-1]` and emits
+/// this config automatically) and asap-query-engine (which reads it back to
+/// actually compute the label at ingest time in
+/// precompute_engine::computed_labels::compute_label_values). Previously the
+/// engine owned this type privately and nothing on the planner side could
+/// construct it - a human had to notice the pattern and hand-write it.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+#[serde(default, deny_unknown_fields)]
+pub struct ComputedLabelConfig {
+    pub r#type: String,
+    pub source_col: String,
+    pub tokenizer: Option<String>,
+    pub filter_regex: Option<String>,
+    pub select: Option<String>,
+    pub on_missing: Option<String>,
+}
+
+impl Default for ComputedLabelConfig {
+    fn default() -> Self {
+        Self {
+            r#type: "field_alias".to_string(),
+            source_col: String::new(),
+            tokenizer: None,
+            filter_regex: None,
+            select: None,
+            on_missing: None,
+        }
+    }
+}

asap-common/dependencies/rs/asap_types/src/lib.rs

@@ -5,17 +5,21 @@ pub mod enums;
 pub mod inference_config;
 pub mod promql_schema;
 pub mod query_config;
+pub mod computed_label;
 pub mod query_requirements;
+pub mod stateful_transition;
 pub mod streaming_config;
 pub mod traits;
 pub mod utils;
 
 pub use aggregation_config::*;
 pub use aggregation_reference::*;
 pub use capability_matching::find_compatible_aggregation;
+pub use computed_label::*;
 pub use enums::*;
 pub use inference_config::*;
 pub use promql_schema::*;
 pub use query_config::*;
 pub use query_requirements::*;
+pub use stateful_transition::*;
 pub use streaming_config::*;

asap-common/dependencies/rs/asap_types/src/stateful_transition.rs

@@ -0,0 +1,48 @@
+use serde::{Deserialize, Serialize};
+
+/// Config for a stateful transition operator: remembers the last value of
+/// `state_column` per `partition_by` key and emits a derived event into
+/// `metric_name` when `predicate` (comparing the remembered previous value
+/// to the current row) holds. This is how patterns like ClickHouse's
+/// `lagInFrame(col) OVER (PARTITION BY ... ORDER BY timestamp)` get lowered
+/// into something the precompute engine's ordinary aggregators can count,
+/// without the planner or query engine ever needing to understand window
+/// functions.
+///
+/// Shared between asap-planner-rs (which detects the SQL pattern and emits
+/// this config into streaming_config.yaml) and asap-query-engine (which
+/// reads it back out to drive the ingest-time operator). Previously these
+/// were two disconnected things: the engine had this struct privately and
+/// nothing ever auto-populated it, so it had to be hand-written per query.
+#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
+#[serde(default, deny_unknown_fields)]
+pub struct StatefulTransitionConfig {
+    /// Generated metric name for the derived event stream.
+    /// Example: derived_lag_transition_path_changes
+    pub metric_name: String,
+
+    /// Columns that define independent state machines.
+    /// Example: [prefix, collector, peer_ip]
+    pub partition_by: Vec<String>,
+
+    /// Column whose previous value is remembered.
+    /// Example: as_path
+    pub state_column: String,
+
+    /// Alias used by the SQL query for the previous value.
+    /// Example: previous_path
+    pub previous_alias: String,
+
+    /// Raw predicate from countIf(...).
+    /// V0 supports AND of simple comparisons:
+    ///   previous_alias != ''
+    ///   previous_alias != state_column
+    ///   previous_alias = 'literal'
+    ///   previous_alias != 'literal'
+    pub predicate: String,
+
+    /// Labels to put on emitted derived samples.
+    /// Usually this is the outer GROUP BY list.
+    /// Empty means global aggregate.
+    pub emit_labels: Vec<String>,
+}

asap-common/dependencies/rs/asap_types/src/streaming_config.rs

@@ -8,19 +8,59 @@ use std::ops::Index;
 
 use crate::aggregation_config::{AggregationConfig, AggregationIdInfo};
 use crate::capability_matching::find_compatible_aggregation as common_find_compatible;
+use crate::computed_label::ComputedLabelConfig;
 use crate::enums::QueryLanguage;
 use crate::inference_config::{InferenceConfig, SchemaConfig};
 use crate::query_requirements::QueryRequirements;
+use crate::stateful_transition::StatefulTransitionConfig;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct StreamingConfig {
     pub aggregation_configs: HashMap<u64, AggregationConfig>,
+    /// Stateful-transition operators (e.g. lagInFrame-derived path-change
+    /// streams) the planner detected while building this config. The
+    /// precompute engine's CSV/HTTP ingest reads these to know what derived
+    /// event streams it needs to maintain before any aggregation in
+    /// `aggregation_configs` can be satisfied against them.
+    #[serde(default)]
+    pub stateful_transitions: Vec<StatefulTransitionConfig>,
+    /// Computed labels (e.g. origin-ASN extraction from as_path) the planner
+    /// detected while building this config. Keyed by the label name the rest
+    /// of the plan (grouping labels, spatial filters) refers to as if it were
+    /// an ordinary column.
+    #[serde(default)]
+    pub computed_label_cols: HashMap<String, ComputedLabelConfig>,
 }
 
 impl StreamingConfig {
     pub fn new(aggregation_configs: HashMap<u64, AggregationConfig>) -> Self {
         Self {
             aggregation_configs,
+            stateful_transitions: Vec::new(),
+            computed_label_cols: HashMap::new(),
+        }
+    }
+
+    pub fn with_stateful_transitions(
+        aggregation_configs: HashMap<u64, AggregationConfig>,
+        stateful_transitions: Vec<StatefulTransitionConfig>,
+    ) -> Self {
+        Self {
+            aggregation_configs,
+            stateful_transitions,
+            computed_label_cols: HashMap::new(),
+        }
+    }
+
+    pub fn with_extras(
+        aggregation_configs: HashMap<u64, AggregationConfig>,
+        stateful_transitions: Vec<StatefulTransitionConfig>,
+        computed_label_cols: HashMap<String, ComputedLabelConfig>,
+    ) -> Self {
+        Self {
+            aggregation_configs,
+            stateful_transitions,
+            computed_label_cols,
         }
     }
 
@@ -101,7 +141,23 @@ impl StreamingConfig {
             }
         }
 
-        Ok(Self::new(aggregation_configs))
+        let stateful_transitions: Vec<StatefulTransitionConfig> = data
+            .get("stateful_transitions")
+            .map(|v| serde_yaml::from_value(v.clone()))
+            .transpose()?
+            .unwrap_or_default();
+
+        let computed_label_cols: HashMap<String, ComputedLabelConfig> = data
+            .get("computed_label_cols")
+            .map(|v| serde_yaml::from_value(v.clone()))
+            .transpose()?
+            .unwrap_or_default();
+
+        Ok(Self::with_extras(
+            aggregation_configs,
+            stateful_transitions,
+            computed_label_cols,
+        ))
     }
 }
 

asap-common/dependencies/rs/sql_utilities/src/ast_matching/mod.rs

@@ -1,8 +1,10 @@
+pub mod pattern_rewrites;
 pub mod sqlhelper;
 pub mod sqlparser_test;
 pub mod sqlpattern_matcher;
 pub mod sqlpattern_parser;
 
+pub use pattern_rewrites::*;
 pub use sqlhelper::{detect_sql_topk, SQLSchema, SqlTopk, Table, TopkWeighting};
 pub use sqlpattern_matcher::*;
 pub use sqlpattern_parser::*;