Merge pull request #119 from caibirdme/dev

feat: support ck level case insensitive

Author: caibirdme

Cargo.lock

@@ -55,7 +55,7 @@ prost = { version = "0.13.1" }
 reqwest = { version = "0.12.7", features = ["json", "native-tls-vendored", "gzip"], default-features = false }
 reqwest-middleware = "0.3.3"
 serde = { version = "1.0.209", features = ["derive"] }
-serde_json = { version = "1.0.125" }
+serde_json = { version = "1.0.127" }
 serde_with = { version = "3.9.0", features = ["json"] }
 sqlbuilder = { path = "sqlbuilder" }
 thiserror = { version = "1.0.63" }

Cargo.toml

@@ -20,6 +20,7 @@ log_source:
         attributes: ["quantity", "code.function"]
       replace_dash_to_dot: true
       default_log_level: debug
+      level_case_sensitive: false
 
   # quickwit:
   #   domain: http://127.0.0.1:7280

config.yaml

@@ -57,6 +57,25 @@ impl IRVisitor for DefaultIRVisitor {
 				cmp: Cmp::Equal(PlaceValue::String(p.value.to_string())),
 			};
 		}
+		if matches!(p.label.to_lowercase().as_str(), "level" | "severitytext") {
+			return Condition {
+				column: Column::Level,
+				cmp: match p.op {
+					Operator::NotEqual => {
+						Cmp::NotEqual(PlaceValue::String(p.value.to_string()))
+					}
+					Operator::RegexMatch => {
+						Cmp::RegexMatch(p.value.to_string())
+					}
+					Operator::RegexNotMatch => {
+						Cmp::RegexNotMatch(p.value.to_string())
+					}
+					Operator::Equal => {
+						Cmp::Equal(PlaceValue::String(p.value.to_string()))
+					}
+				},
+			};
+		}
 		Condition {
 			column: maybe_nested_key(&p.label),
 			cmp: match p.op {

sqlbuilder/src/visit.rs

@@ -90,6 +90,7 @@ pub struct ClickhouseLog {
 	pub replace_dash_to_dot: Option<bool>,
 	#[serde(default = "default_log_level")]
 	pub default_log_level: String,
+	pub level_case_sensitive: Option<bool>,
 }
 
 fn default_log_level() -> String {
@@ -228,6 +229,7 @@ mod tests {
 			},
 			replace_dash_to_dot: None,
 			default_log_level: "info".to_string(),
+			level_case_sensitive: None,
 		});
 		assert_eq!(expect, actual);
 	}
@@ -289,6 +291,7 @@ mod tests {
 			},
 			replace_dash_to_dot: Some(true),
 			default_log_level: "debug".to_string(),
+			level_case_sensitive: Some(false),
 		};
 		assert_eq!(
 			cfg.log_source,

src/config.rs

@@ -6,19 +6,31 @@ use sqlbuilder::builder::*;
 pub struct CKLogConverter<T: TableSchema> {
 	table: T,
 	replace_dash_to_dot: bool,
+	level_insenstive: bool,
 }
 
 impl<T: TableSchema> CKLogConverter<T> {
-	pub fn new(table: T, replace_dash_to_dot: bool) -> Self {
+	pub fn new(
+		table: T,
+		replace_dash_to_dot: bool,
+		level_insenstive: bool,
+	) -> Self {
 		Self {
 			table,
 			replace_dash_to_dot,
+			level_insenstive,
 		}
 	}
 }
 
 impl<T: TableSchema> QueryConverter for CKLogConverter<T> {
 	fn convert_condition(&self, c: &Condition) -> String {
+		// special case for level
+		if matches!(c.column, Column::Level) {
+			if let Some(s) = self.convert_level(&c.cmp) {
+				return s;
+			}
+		}
 		let col_name = self.column_name(&c.column);
 		match &c.cmp {
 			Cmp::Equal(v) => format!("{} = {}", col_name, v),
@@ -62,6 +74,31 @@ impl<T: TableSchema> QueryConverter for CKLogConverter<T> {
 }
 
 impl<T: TableSchema> CKLogConverter<T> {
+	fn convert_level(&self, cmp: &Cmp) -> Option<String> {
+		let insensitive = self.level_insenstive;
+		let key = self.table.level_key();
+		match cmp {
+			Cmp::Equal(v) => {
+				if insensitive {
+					Some(format!("{} ILIKE {}", key, v))
+				} else {
+					Some(format!("{} = {}", key, v))
+				}
+			}
+			Cmp::NotEqual(v) => {
+				if insensitive {
+					Some(format!("{} NOT ILIKE {}", key, v))
+				} else {
+					Some(format!("{} != {}", key, v))
+				}
+			}
+			Cmp::RegexMatch(v) => Some(format!("match({}, '{}')", key, v)),
+			Cmp::RegexNotMatch(v) => {
+				Some(format!("NOT match({}, '{}')", key, v))
+			}
+			_ => None,
+		}
+	}
 	fn column_name(&self, c: &Column) -> String {
 		match c {
 			Column::Message => self.table.msg_key().to_string(),

src/storage/ck/converter.rs

@@ -8,7 +8,7 @@ use logql::parser::{LogQuery, MetricQuery};
 use reqwest::Client;
 use serde_json::Value as JSONValue;
 use sqlbuilder::{
-	builder::{time_range_into_timing, QueryPlan, TableSchema},
+	builder::{time_range_into_timing, QueryConverter, QueryPlan, TableSchema},
 	visit::{DefaultIRVisitor, LogQLVisitor},
 };
 use std::{
@@ -48,6 +48,13 @@ impl CKLogQuerier {
 			tx,
 		}
 	}
+	fn new_converter(&self) -> CKLogConverter<LogTable> {
+		CKLogConverter::new(
+			self.schema.clone(),
+			self.ck_cfg.replace_dash_to_dot.unwrap_or(false),
+			!self.ck_cfg.level_case_sensitive.unwrap_or(false),
+		)
+	}
 }
 
 #[async_trait]
@@ -57,12 +64,7 @@ impl LogStorage for CKLogQuerier {
 		q: &LogQuery,
 		opt: QueryLimits,
 	) -> Result<Vec<LogItem>> {
-		let sql = logql_to_sql(
-			q,
-			opt,
-			&self.schema,
-			self.ck_cfg.replace_dash_to_dot.unwrap_or(false),
-		);
+		let sql = logql_to_sql(q, opt, &self.schema, self.new_converter());
 		let mut results = vec![];
 		let rows =
 			send_query(self.cli.clone(), self.ck_cfg.common.clone(), sql)
@@ -86,7 +88,12 @@ impl LogStorage for CKLogQuerier {
 		q: &MetricQuery,
 		opt: QueryLimits,
 	) -> Result<Vec<MetricItem>> {
-		let sql = new_from_metricquery(q, opt, self.schema.clone());
+		let sql = new_from_metricquery(
+			q,
+			opt,
+			self.schema.clone(),
+			self.new_converter(),
+		);
 		let mut results = vec![];
 		let rows =
 			send_query(self.cli.clone(), self.ck_cfg.common.clone(), sql)
@@ -274,12 +281,13 @@ fn new_from_metricquery(
 	q: &MetricQuery,
 	limits: QueryLimits,
 	schema: LogTable,
+	converter: impl QueryConverter,
 ) -> String {
 	let v = LogQLVisitor::new(DefaultIRVisitor {});
 	let selection = v.visit(&q.log_query);
 	let step = limits.step.unwrap_or(DEFAULT_STEP);
 	let qp = QueryPlan::new(
-		CKLogConverter::new(schema.clone(), false),
+		converter,
 		schema.clone(),
 		vec![
 			to_start_interval(step).to_string(),
@@ -299,12 +307,12 @@ fn logql_to_sql(
 	q: &LogQuery,
 	limits: QueryLimits,
 	schema: &LogTable,
-	replace_dash: bool,
+	converter: impl QueryConverter,
 ) -> String {
 	let v = LogQLVisitor::new(DefaultIRVisitor {});
 	let selection = v.visit(q);
 	let qp = QueryPlan::new(
-		CKLogConverter::new(schema.clone(), replace_dash),
+		converter,
 		schema.clone(),
 		schema.projection(),
 		selection,
@@ -435,7 +443,7 @@ impl TableSchema for LogTable {
 		self.table.as_str()
 	}
 	fn level_key(&self) -> &str {
-		"SeverityNumber"
+		"SeverityText"
 	}
 	fn trace_key(&self) -> &str {
 		"TraceId"

src/storage/ck/log.rs

@@ -73,7 +73,8 @@ impl TraceStorage for CKTraceQuerier {
 				return Ok(vec![]);
 			}
 			Expression::SpanSet(sp) => {
-				let converter = CKLogConverter::new(self.schema.clone(), true);
+				let converter =
+					CKLogConverter::new(self.schema.clone(), true, true);
 				let sql = single_spanset_query(
 					sp,
 					self.schema.clone(),
@@ -413,7 +414,7 @@ mod tests {
 		for (name, tc) in cases {
 			let expr = parse_traceql(&tc.input).unwrap();
 			if let Expression::SpanSet(sp) = expr {
-				let converter = CKLogConverter::new(schema.clone(), true);
+				let converter = CKLogConverter::new(schema.clone(), true, true);
 				let sql = single_spanset_query(
 					&sp,
 					schema.clone(),