Skip to content

Commit 5a22b3c

Browse files
refactor: change magic strings to enums (#279)
* Converted window type and cleanup policies to enums * added more enums * Changed accumulator strings to enums
1 parent 465f69f commit 5a22b3c

63 files changed

Lines changed: 1421 additions & 790 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎asap-common/dependencies/rs/asap_types/src/aggregation_config.rs‎

Lines changed: 21 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -3,25 +3,26 @@ use serde_json::Value;
33
use serde_yaml;
44
use std::collections::HashMap;
55

6-
use crate::enums::QueryLanguage;
6+
use crate::enums::{QueryLanguage, WindowType};
77
use crate::traits::SerializableToSink;
88
use crate::utils::normalize_spatial_filter;
99
use promql_utilities::data_model::KeyByLabelNames;
10+
use promql_utilities::query_logics::enums::AggregationType;
1011

1112
#[derive(Debug, Clone, Serialize, Deserialize)]
1213
pub struct AggregationConfig {
1314
pub aggregation_id: u64,
14-
pub aggregation_type: String,
15+
pub aggregation_type: AggregationType,
1516
pub aggregation_sub_type: String,
1617
pub parameters: HashMap<String, Value>,
1718
pub grouping_labels: KeyByLabelNames,
1819
pub aggregated_labels: KeyByLabelNames,
1920
pub rollup_labels: KeyByLabelNames,
2021
pub original_yaml: String,
2122

22-
pub window_size: u64, // Window size in seconds (e.g., 900s for 15m)
23-
pub slide_interval: u64, // Slide/hop interval in seconds (e.g., 30s)
24-
pub window_type: String, // "tumbling" or "sliding"
23+
pub window_size: u64, // Window size in seconds (e.g., 900s for 15m)
24+
pub slide_interval: u64, // Slide/hop interval in seconds (e.g., 30s)
25+
pub window_type: WindowType, // Tumbling or Sliding
2526

2627
pub spatial_filter: String,
2728
pub spatial_filter_normalized: String,
@@ -41,8 +42,8 @@ pub struct AggregationConfig {
4142
pub struct AggregationIdInfo {
4243
pub aggregation_id_for_key: u64,
4344
pub aggregation_id_for_value: u64,
44-
pub aggregation_type_for_key: String,
45-
pub aggregation_type_for_value: String,
45+
pub aggregation_type_for_key: AggregationType,
46+
pub aggregation_type_for_value: AggregationType,
4647
}
4748

4849
// TODO: need to implement deserialization methods
@@ -51,7 +52,7 @@ impl AggregationConfig {
5152
#[allow(clippy::too_many_arguments)]
5253
pub fn new(
5354
aggregation_id: u64,
54-
aggregation_type: String,
55+
aggregation_type: AggregationType,
5556
aggregation_sub_type: String,
5657
parameters: HashMap<String, Value>,
5758
grouping_labels: KeyByLabelNames,
@@ -60,7 +61,7 @@ impl AggregationConfig {
6061
original_yaml: String,
6162
window_size: u64,
6263
slide_interval: u64,
63-
window_type: String,
64+
window_type: WindowType,
6465
spatial_filter: String,
6566
metric: String,
6667
num_aggregates_to_retain: Option<u64>,
@@ -116,10 +117,11 @@ impl AggregationConfig {
116117
.as_u64()
117118
.ok_or("Missing aggregationId")?;
118119

119-
let aggregation_type = data["aggregationType"]
120+
let aggregation_type: AggregationType = data["aggregationType"]
120121
.as_str()
121122
.ok_or("Missing aggregationType")?
122-
.to_string();
123+
.parse()
124+
.map_err(|e: String| e)?;
123125

124126
let aggregation_sub_type = data["aggregationSubType"]
125127
.as_str()
@@ -148,7 +150,8 @@ impl AggregationConfig {
148150
.get("windowType")
149151
.and_then(|v| v.as_str())
150152
.unwrap_or("tumbling")
151-
.to_string();
153+
.parse::<WindowType>()
154+
.unwrap_or_default();
152155

153156
let slide_interval = data
154157
.get("slideInterval")
@@ -240,10 +243,11 @@ impl AggregationConfig {
240243
.collect(),
241244
);
242245

243-
let aggregation_type = aggregation_data["aggregationType"]
246+
let aggregation_type: AggregationType = aggregation_data["aggregationType"]
244247
.as_str()
245248
.ok_or_else(|| anyhow::anyhow!("Missing aggregationType"))?
246-
.to_string();
249+
.parse()
250+
.map_err(|e: String| anyhow::anyhow!(e))?;
247251

248252
let aggregation_sub_type = aggregation_data["aggregationSubType"]
249253
.as_str()
@@ -270,7 +274,8 @@ impl AggregationConfig {
270274
.get("windowType")
271275
.and_then(|v| v.as_str())
272276
.unwrap_or("tumbling")
273-
.to_string();
277+
.parse::<WindowType>()
278+
.unwrap_or_default();
274279

275280
let slide_interval = aggregation_data
276281
.get("slideInterval")
@@ -348,7 +353,7 @@ impl SerializableToSink for AggregationConfig {
348353
"originalYaml": self.original_yaml,
349354
"windowSize": self.window_size,
350355
"slideInterval": self.slide_interval,
351-
"windowType": self.window_type,
356+
"windowType": self.window_type.to_string(),
352357
"spatialFilter": self.spatial_filter,
353358
"metric": self.metric,
354359
});

‎asap-common/dependencies/rs/asap_types/src/capability_matching.rs‎

Lines changed: 32 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -6,31 +6,35 @@ use promql_utilities::query_logics::enums::Statistic;
66
use tracing::{debug, warn};
77

88
use crate::aggregation_config::{AggregationConfig, AggregationIdInfo};
9+
use crate::enums::WindowType;
910
use crate::query_requirements::QueryRequirements;
1011
use crate::utils::normalize_spatial_filter;
12+
use promql_utilities::query_logics::enums::AggregationType;
1113

1214
// ---------------------------------------------------------------------------
1315
// Pure compatibility helpers
1416
// ---------------------------------------------------------------------------
1517

16-
/// Returns the aggregation_type strings that can serve this statistic.
17-
pub fn compatible_agg_types(stat: Statistic) -> &'static [&'static str] {
18+
/// Returns the aggregation types that can serve this statistic.
19+
pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] {
1820
match stat {
19-
Statistic::Sum => &["Sum", "MultipleSumAccumulator"],
21+
Statistic::Sum => &[AggregationType::Sum, AggregationType::MultipleSum],
2022
Statistic::Count => &[
21-
"CountMinSketch",
22-
"CountMinSketchWithHeap",
23-
"CountMinSketchWithHeapAccumulator",
23+
AggregationType::CountMinSketch,
24+
AggregationType::CountMinSketchWithHeap,
2425
],
25-
Statistic::Min => &["MinMax", "MultipleMinMaxAccumulator"],
26-
Statistic::Max => &["MinMax", "MultipleMinMaxAccumulator"],
27-
Statistic::Quantile => &["DatasketchesKLL", "HydraKLL"],
28-
Statistic::Rate | Statistic::Increase => &["Increase", "MultipleIncreaseAccumulator"],
29-
Statistic::Cardinality => &["SetAggregator", "DeltaSetAggregator"],
30-
Statistic::Topk => &[
31-
"CountMinSketchWithHeap",
32-
"CountMinSketchWithHeapAccumulator",
26+
Statistic::Min | Statistic::Max => {
27+
&[AggregationType::MinMax, AggregationType::MultipleMinMax]
28+
}
29+
Statistic::Quantile => &[AggregationType::DatasketchesKLL, AggregationType::HydraKLL],
30+
Statistic::Rate | Statistic::Increase => {
31+
&[AggregationType::Increase, AggregationType::MultipleIncrease]
32+
}
33+
Statistic::Cardinality => &[
34+
AggregationType::SetAggregator,
35+
AggregationType::DeltaSetAggregator,
3336
],
37+
Statistic::Topk => &[AggregationType::CountMinSketchWithHeap],
3438
}
3539
}
3640

@@ -46,20 +50,13 @@ pub fn required_sub_type(stat: Statistic) -> Option<&'static str> {
4650

4751
/// Whether this value aggregation type requires a paired key aggregation
4852
/// (`SetAggregator` or `DeltaSetAggregator`).
49-
pub fn is_multi_population_value_type(agg_type: &str) -> bool {
50-
matches!(
51-
agg_type,
52-
"MultipleSumAccumulator"
53-
| "MultipleMinMaxAccumulator"
54-
| "MultipleIncreaseAccumulator"
55-
| "CountMinSketchWithHeap"
56-
| "CountMinSketchWithHeapAccumulator"
57-
)
53+
pub fn is_multi_population_value_type(agg_type: AggregationType) -> bool {
54+
agg_type.is_multi_population_value_type()
5855
}
5956

6057
/// Whether this type is a key aggregation (tracks which label-value combinations exist).
61-
fn is_key_agg_type(agg_type: &str) -> bool {
62-
matches!(agg_type, "SetAggregator" | "DeltaSetAggregator")
58+
fn is_key_agg_type(agg_type: AggregationType) -> bool {
59+
agg_type.is_key_agg_type()
6360
}
6461

6562
/// Window compatibility: can `config` serve a query needing `data_range_ms`?
@@ -76,9 +73,9 @@ pub fn window_compatible(config: &AggregationConfig, data_range_ms: Option<u64>)
7673
if window_ms == 0 || range == 0 {
7774
return false;
7875
}
79-
match config.window_type.as_str() {
80-
"sliding" => range == window_ms,
81-
_ => range % window_ms == 0, // tumbling (or unknown — treat as tumbling)
76+
match config.window_type {
77+
WindowType::Sliding => range == window_ms,
78+
WindowType::Tumbling => range % window_ms == 0,
8279
}
8380
}
8481

@@ -152,7 +149,7 @@ pub fn find_compatible_aggregation(
152149
.values()
153150
.filter(|c| {
154151
let ok = c.metric == requirements.metric
155-
&& types.contains(&c.aggregation_type.as_str())
152+
&& types.contains(&c.aggregation_type)
156153
&& sub_type.is_none_or(|st| c.aggregation_sub_type == st)
157154
&& window_compatible(c, requirements.data_range_ms)
158155
&& labels_compatible(&c.grouping_labels, &requirements.grouping_labels)
@@ -218,11 +215,11 @@ pub fn find_compatible_aggregation(
218215
}
219216

220217
// If value type is multi-population, find the paired key aggregation.
221-
let key_agg: &AggregationConfig = if is_multi_population_value_type(&value_agg.aggregation_type)
218+
let key_agg: &AggregationConfig = if is_multi_population_value_type(value_agg.aggregation_type)
222219
{
223220
let ka = configs
224221
.values()
225-
.find(|c| c.metric == requirements.metric && is_key_agg_type(&c.aggregation_type));
222+
.find(|c| c.metric == requirements.metric && is_key_agg_type(c.aggregation_type));
226223
if ka.is_none() {
227224
warn!(
228225
metric = %requirements.metric,
@@ -246,9 +243,9 @@ pub fn find_compatible_aggregation(
246243

247244
Some(AggregationIdInfo {
248245
aggregation_id_for_value: value_agg.aggregation_id,
249-
aggregation_type_for_value: value_agg.aggregation_type.clone(),
246+
aggregation_type_for_value: value_agg.aggregation_type,
250247
aggregation_id_for_key: key_agg.aggregation_id,
251-
aggregation_type_for_key: key_agg.aggregation_type.clone(),
248+
aggregation_type_for_key: key_agg.aggregation_type,
252249
})
253250
}
254251

@@ -279,7 +276,7 @@ mod tests {
279276
let spatial_filter_normalized = normalize_spatial_filter(spatial_filter);
280277
AggregationConfig {
281278
aggregation_id: id,
282-
aggregation_type: agg_type.to_string(),
279+
aggregation_type: agg_type.parse::<AggregationType>().expect("valid agg type"),
283280
aggregation_sub_type: sub_type.to_string(),
284281
parameters: HashMap::new(),
285282
grouping_labels,
@@ -288,7 +285,7 @@ mod tests {
288285
original_yaml: String::new(),
289286
window_size: window_size_s,
290287
slide_interval: window_size_s,
291-
window_type: window_type.to_string(),
288+
window_type: window_type.parse::<WindowType>().unwrap_or_default(),
292289
spatial_filter: spatial_filter.to_string(),
293290
spatial_filter_normalized,
294291
metric: metric.to_string(),

‎asap-common/dependencies/rs/asap_types/src/enums.rs‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,9 @@
1+
use std::fmt;
2+
use std::str::FromStr;
3+
4+
// Re-export AggregationType from promql_utilities (defined there to avoid circular deps).
5+
pub use promql_utilities::query_logics::enums::AggregationType;
6+
17
#[derive(clap::ValueEnum, Clone, Copy, Debug, PartialEq)]
28
#[allow(non_camel_case_types)]
39
pub enum QueryLanguage {
@@ -23,3 +29,58 @@ pub enum CleanupPolicy {
2329
/// Never clean up aggregates
2430
NoCleanup,
2531
}
32+
33+
impl fmt::Display for CleanupPolicy {
34+
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
35+
match self {
36+
CleanupPolicy::CircularBuffer => write!(f, "circular_buffer"),
37+
CleanupPolicy::ReadBased => write!(f, "read_based"),
38+
CleanupPolicy::NoCleanup => write!(f, "no_cleanup"),
39+
}
40+
}
41+
}
42+
43+
impl FromStr for CleanupPolicy {
44+
type Err = String;
45+
46+
fn from_str(s: &str) -> Result<Self, Self::Err> {
47+
match s {
48+
"circular_buffer" => Ok(CleanupPolicy::CircularBuffer),
49+
"read_based" => Ok(CleanupPolicy::ReadBased),
50+
"no_cleanup" => Ok(CleanupPolicy::NoCleanup),
51+
_ => Err(format!("Unknown cleanup policy: '{s}'")),
52+
}
53+
}
54+
}
55+
56+
/// Window type for streaming aggregations.
57+
#[derive(
58+
Clone, Debug, Copy, Default, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize,
59+
)]
60+
#[serde(rename_all = "snake_case")]
61+
pub enum WindowType {
62+
#[default]
63+
Tumbling,
64+
Sliding,
65+
}
66+
67+
impl fmt::Display for WindowType {
68+
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
69+
match self {
70+
WindowType::Tumbling => write!(f, "tumbling"),
71+
WindowType::Sliding => write!(f, "sliding"),
72+
}
73+
}
74+
}
75+
76+
impl FromStr for WindowType {
77+
type Err = String;
78+
79+
fn from_str(s: &str) -> Result<Self, Self::Err> {
80+
match s.to_lowercase().as_str() {
81+
"tumbling" => Ok(WindowType::Tumbling),
82+
"sliding" => Ok(WindowType::Sliding),
83+
_ => Err(format!("Unknown window type: '{s}'")),
84+
}
85+
}
86+
}

‎asap-common/dependencies/rs/asap_types/src/inference_config.rs‎

Lines changed: 4 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -172,15 +172,12 @@ impl InferenceConfig {
172172
)
173173
})?;
174174

175-
match name {
176-
"circular_buffer" => Ok(CleanupPolicy::CircularBuffer),
177-
"read_based" => Ok(CleanupPolicy::ReadBased),
178-
"no_cleanup" => Ok(CleanupPolicy::NoCleanup),
179-
_ => Err(anyhow::anyhow!(
175+
name.parse::<CleanupPolicy>().map_err(|_| {
176+
anyhow::anyhow!(
180177
"Invalid cleanup policy: '{}'. Valid options: circular_buffer, read_based, no_cleanup",
181178
name
182-
)),
183-
}
179+
)
180+
})
184181
}
185182

186183
fn parse_query_configs(

‎asap-common/dependencies/rs/promql_utilities/src/ast_matching/promql_pattern.rs‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -448,11 +448,11 @@ impl PromQLPattern {
448448
debug!("Collecting aggregation token as: {}", collect_as);
449449
let modifier = match &agg.modifier {
450450
Some(LabelModifier::Include(labels)) => Some(AggregationModifier {
451-
modifier_type: "by".to_string(),
451+
modifier_type: AggregationModifierType::By,
452452
labels: labels.labels.clone(),
453453
}),
454454
Some(LabelModifier::Exclude(labels)) => Some(AggregationModifier {
455-
modifier_type: "without".to_string(),
455+
modifier_type: AggregationModifierType::Without,
456456
labels: labels.labels.clone(),
457457
}),
458458
None => None,
@@ -844,16 +844,24 @@ impl Default for PromQLMatchResult {
844844
}
845845
}
846846

847+
/// Whether a PromQL aggregation modifier is `by` or `without`.
848+
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)]
849+
#[serde(rename_all = "lowercase")]
850+
pub enum AggregationModifierType {
851+
By,
852+
Without,
853+
}
854+
847855
/// Represents aggregation modifiers like "by" or "without"
848856
#[derive(Debug, Clone, Serialize)]
849857
pub struct AggregationModifier {
850-
pub modifier_type: String, // "by" or "without"
858+
pub modifier_type: AggregationModifierType,
851859
pub labels: Vec<String>,
852860
}
853861

854862
impl AggregationModifier {
855863
/// Create a new AggregationModifier
856-
pub fn new(modifier_type: String, labels: Vec<String>) -> Self {
864+
pub fn new(modifier_type: AggregationModifierType, labels: Vec<String>) -> Self {
857865
Self {
858866
modifier_type,
859867
labels,

0 commit comments

Comments
 (0)