From ea5146762af4d1f6987dd1b4d9509471209b3448 Mon Sep 17 00:00:00 2001 From: Ruihang Xia Date: Thu, 2 Feb 2023 15:31:41 +0800 Subject: [PATCH] chore(deps): bump promql-parser (#929) * fix promql crate Signed-off-by: Ruihang Xia * migrate to new api Signed-off-by: Ruihang Xia * fix aggregator test Signed-off-by: Ruihang Xia * fix styles Signed-off-by: Ruihang Xia --------- Signed-off-by: Ruihang Xia --- Cargo.lock | 2 +- src/common/error/src/ext.rs | 39 ++++ src/promql/Cargo.toml | 2 +- src/promql/src/error.rs | 6 +- src/promql/src/extension_plan/normalize.rs | 5 +- src/promql/src/planner.rs | 260 +++++++++++++-------- src/query/Cargo.toml | 2 +- src/query/src/parser.rs | 40 +--- 8 files changed, 213 insertions(+), 143 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a01b4449491..21eaf6c62eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5225,7 +5225,7 @@ dependencies = [ [[package]] name = "promql-parser" version = "0.0.1" -source = "git+https://github.com/GreptimeTeam/promql-parser.git?rev=fec3c8bcee982b8add2a77d07818d3cbe92b89fe#fec3c8bcee982b8add2a77d07818d3cbe92b89fe" +source = "git+https://github.com/GreptimeTeam/promql-parser.git?rev=d027ce428a6a2df5a652b8558608c77d33c31644#d027ce428a6a2df5a652b8558608c77d33c31644" dependencies = [ "cfgrammar", "lazy_static", diff --git a/src/common/error/src/ext.rs b/src/common/error/src/ext.rs index 573411e5359..371c3434daf 100644 --- a/src/common/error/src/ext.rs +++ b/src/common/error/src/ext.rs @@ -88,6 +88,45 @@ impl crate::snafu::ErrorCompat for BoxedError { } } +/// Error type with plain error message +#[derive(Debug)] +pub struct PlainError { + msg: String, + status_code: StatusCode, +} + +impl PlainError { + pub fn new(msg: String, status_code: StatusCode) -> Self { + Self { msg, status_code } + } +} + +impl std::fmt::Display for PlainError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.msg) + } +} + +impl std::error::Error for PlainError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + None + } +} + +impl crate::ext::ErrorExt for PlainError { + fn status_code(&self) -> crate::status_code::StatusCode { + self.status_code + } + + fn backtrace_opt(&self) -> Option<&crate::snafu::Backtrace> { + None + } + + fn as_any(&self) -> &dyn std::any::Any { + self as _ + } +} + #[cfg(test)] mod tests { use std::error::Error; diff --git a/src/promql/Cargo.toml b/src/promql/Cargo.toml index 8a8879127da..45d68ca08f4 100644 --- a/src/promql/Cargo.toml +++ b/src/promql/Cargo.toml @@ -13,7 +13,7 @@ common-catalog = { path = "../common/catalog" } datafusion.workspace = true datatypes = { path = "../datatypes" } futures = "0.3" -promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "fec3c8bcee982b8add2a77d07818d3cbe92b89fe" } +promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "d027ce428a6a2df5a652b8558608c77d33c31644" } session = { path = "../session" } snafu = { version = "0.7", features = ["backtraces"] } table = { path = "../table" } diff --git a/src/promql/src/error.rs b/src/promql/src/error.rs index de6a7285d71..a42c4b2d3e0 100644 --- a/src/promql/src/error.rs +++ b/src/promql/src/error.rs @@ -16,6 +16,7 @@ use std::any::Any; use common_error::prelude::*; use datafusion::error::DataFusionError; +use promql_parser::label::Label; use promql_parser::parser::{Expr as PromExpr, TokenType}; #[derive(Debug, Snafu)] @@ -48,10 +49,11 @@ pub enum Error { #[snafu(display("Cannot find value columns in table {}", table))] ValueNotFound { table: String, backtrace: Backtrace }, - #[snafu(display("Cannot find label in table {}, source: {}", table, source))] + #[snafu(display("Cannot find label {} in table {}", label, table,))] LabelNotFound { table: String, - source: datafusion::error::DataFusionError, + label: Label, + backtrace: Backtrace, }, #[snafu(display("Cannot find the table {}", table))] diff --git a/src/promql/src/extension_plan/normalize.rs b/src/promql/src/extension_plan/normalize.rs index 78e30f1e8db..34b84d0c9f6 100644 --- a/src/promql/src/extension_plan/normalize.rs +++ b/src/promql/src/extension_plan/normalize.rs @@ -16,7 +16,6 @@ use std::any::Any; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use std::time::Duration; use datafusion::arrow::compute; use datafusion::common::{DFSchemaRef, Result as DataFusionResult, Statistics}; @@ -91,12 +90,12 @@ impl UserDefinedLogicalNode for SeriesNormalize { impl SeriesNormalize { pub fn new>( - offset: Duration, + offset: Millisecond, time_index_column_name: N, input: LogicalPlan, ) -> Self { Self { - offset: offset.as_millis() as i64, + offset, time_index_column_name: time_index_column_name.as_ref().to_string(), input, } diff --git a/src/promql/src/planner.rs b/src/promql/src/planner.rs index ace1e41c9e5..c8d22139290 100644 --- a/src/promql/src/planner.rs +++ b/src/promql/src/planner.rs @@ -15,8 +15,9 @@ use std::collections::HashSet; use std::str::FromStr; use std::sync::Arc; -use std::time::{Duration, UNIX_EPOCH}; +use std::time::UNIX_EPOCH; +use datafusion::common::DFSchemaRef; use datafusion::datasource::DefaultTableSource; use datafusion::logical_expr::expr::AggregateFunction; use datafusion::logical_expr::{ @@ -30,17 +31,17 @@ use datafusion::sql::planner::ContextProvider; use datafusion::sql::TableReference; use promql_parser::label::{MatchOp, Matchers, METRIC_NAME}; use promql_parser::parser::{ - token, AggregateExpr, BinaryExpr as PromBinaryExpr, Call, EvalStmt, Expr as PromExpr, Function, - MatrixSelector, NumberLiteral, ParenExpr, StringLiteral, SubqueryExpr, TokenType, UnaryExpr, - VectorSelector, + token, AggModifier, AggregateExpr, BinaryExpr as PromBinaryExpr, Call, EvalStmt, + Expr as PromExpr, Function, MatrixSelector, NumberLiteral, Offset, ParenExpr, StringLiteral, + SubqueryExpr, TokenType, UnaryExpr, VectorSelector, }; -use snafu::{OptionExt, ResultExt}; +use snafu::{ensure, OptionExt, ResultExt}; use table::table::adapter::DfTableProviderAdapter; use crate::error::{ DataFusionPlanningSnafu, ExpectExprSnafu, LabelNotFoundSnafu, MultipleVectorSnafu, Result, - TableNameNotFoundSnafu, TableNotFoundSnafu, TimeIndexNotFoundSnafu, UnexpectedPlanExprSnafu, - UnexpectedTokenSnafu, UnknownTableSnafu, UnsupportedExprSnafu, ValueNotFoundSnafu, + TableNameNotFoundSnafu, TableNotFoundSnafu, TimeIndexNotFoundSnafu, UnexpectedTokenSnafu, + UnknownTableSnafu, UnsupportedExprSnafu, ValueNotFoundSnafu, }; use crate::extension_plan::{InstantManipulate, Millisecond, RangeManipulate, SeriesNormalize}; @@ -94,39 +95,11 @@ impl PromPlanner { // TODO(ruihang): support param param: _param, grouping, - without, }) => { let input = self.prom_expr_to_plan(*expr.clone())?; // calculate columns to group by - let schema = input.schema(); - let group_columns_indices = grouping - .iter() - .map(|label| { - schema - .index_of_column_by_name(None, label) - .with_context(|_| LabelNotFoundSnafu { - table: self.ctx.table_name.clone().unwrap(), - }) - }) - .collect::>>()?; - let value_names = self.ctx.value_columns.iter().collect::>(); - let group_exprs = schema - .fields() - .iter() - .enumerate() - .filter_map(|(i, field)| { - if *without != group_columns_indices.contains(&i) - && Some(field.name()) != self.ctx.time_index_column.as_ref() - && !value_names.contains(&field.name()) - { - Some(DfExpr::Column(Column::from(field.name()))) - } else { - None - } - }) - .collect::>(); - + let group_exprs = self.agg_modifier_to_col(input.schema(), grouping)?; // convert op and value columns to aggregate exprs let aggr_exprs = self.create_aggregate_exprs(*op)?; @@ -213,12 +186,12 @@ impl PromPlanner { PromExpr::VectorSelector(VectorSelector { name: _, offset, - start_or_end: _, label_matchers, + at: _, }) => { let matchers = self.preprocess_label_matchers(label_matchers)?; self.setup_context()?; - let normalize = self.selector_to_series_normalize_plan(*offset, matchers)?; + let normalize = self.selector_to_series_normalize_plan(offset, matchers)?; let manipulate = InstantManipulate::new( self.ctx.start, self.ctx.end, @@ -238,24 +211,14 @@ impl PromPlanner { vector_selector, range, }) => { - let normalize = match &**vector_selector { - PromExpr::VectorSelector(VectorSelector { - name: _, - offset, - start_or_end: _, - label_matchers, - })=> { - let matchers = self.preprocess_label_matchers(label_matchers)?; - self.setup_context()?; - self.selector_to_series_normalize_plan(*offset, matchers)? - } - _ => UnexpectedPlanExprSnafu { - desc: format!( - "MatrixSelector must contains a VectorSelector, but found {vector_selector:?}", - ), - } - .fail()?, - }; + let VectorSelector { + offset, + label_matchers, + .. + } = vector_selector; + let matchers = self.preprocess_label_matchers(label_matchers)?; + self.setup_context()?; + let normalize = self.selector_to_series_normalize_plan(offset, matchers)?; let manipulate = RangeManipulate::new( self.ctx.start, self.ctx.end, @@ -276,7 +239,7 @@ impl PromPlanner { }) } PromExpr::Call(Call { func, args }) => { - let args = self.create_function_args(args)?; + let args = self.create_function_args(&args.args)?; let input = self.prom_expr_to_plan(args.input.with_context(|| ExpectExprSnafu { expr: prom_expr.clone(), @@ -298,22 +261,21 @@ impl PromPlanner { /// Extract metric name from `__name__` matcher and set it into [PromPlannerContext]. /// Returns a new [Matchers] that doesn't contains metric name matcher. fn preprocess_label_matchers(&mut self, label_matchers: &Matchers) -> Result { - let mut matchers = Vec::with_capacity(label_matchers.matchers.len()); + let mut matchers = HashSet::new(); for matcher in &label_matchers.matchers { // TODO(ruihang): support other metric match ops if matcher.name == METRIC_NAME && matches!(matcher.op, MatchOp::Equal) { self.ctx.table_name = Some(matcher.value.clone()); } else { - matchers.push(matcher.clone()); + matchers.insert(matcher.clone()); } } - Ok(Matchers { matchers }) } fn selector_to_series_normalize_plan( &self, - offset: Option, + offset: &Option, label_matchers: Matchers, ) -> Result { let table_name = self.ctx.table_name.clone().unwrap(); @@ -341,9 +303,13 @@ impl PromPlanner { ); // make series_normalize plan - let offset = offset.unwrap_or_default(); + let offset_duration = match offset { + Some(Offset::Pos(duration)) => duration.as_millis() as Millisecond, + Some(Offset::Neg(duration)) => -(duration.as_millis() as Millisecond), + None => 0, + }; let series_normalize = SeriesNormalize::new( - offset, + offset_duration, self.ctx .time_index_column .clone() @@ -356,6 +322,71 @@ impl PromPlanner { Ok(logical_plan) } + /// Convert [AggModifier] to [Column] exprs for aggregation. + fn agg_modifier_to_col( + &self, + input_schema: &DFSchemaRef, + modifier: &AggModifier, + ) -> Result> { + match modifier { + AggModifier::By(labels) => { + let mut exprs = Vec::with_capacity(labels.len()); + for label in labels { + let field = input_schema + .field_with_unqualified_name(label) + .map_err(|_| { + LabelNotFoundSnafu { + table: self + .ctx + .table_name + .clone() + .unwrap_or("no_table_name".to_string()), + label: label.clone(), + } + .build() + })?; + exprs.push(DfExpr::Column(Column::from(field.name()))); + } + Ok(exprs) + } + AggModifier::Without(labels) => { + let mut all_fields = input_schema + .fields() + .iter() + .map(|f| f.name()) + .collect::>(); + // remove "without"-ed fields + for label in labels { + ensure!( + // ensure this field was existed + all_fields.remove(label), + LabelNotFoundSnafu { + table: self + .ctx + .table_name + .clone() + .unwrap_or("no_table_name".to_string()), + label: label.clone(), + } + ); + } + // remove time index and value fields + if let Some(time_index) = &self.ctx.time_index_column { + all_fields.remove(time_index); + } + for value in &self.ctx.value_columns { + all_fields.remove(value); + } + // collect remaining fields and convert to col expr + let exprs = all_fields + .into_iter() + .map(|c| DfExpr::Column(Column::from(c))) + .collect(); + Ok(exprs) + } + } + } + // TODO(ruihang): ignore `MetricNameLabel` (`__name__`) matcher fn matchers_to_expr(&self, label_matchers: Matchers) -> Result> { let mut exprs = Vec::with_capacity(label_matchers.matchers.len()); @@ -666,7 +697,7 @@ struct FunctionArgs { #[cfg(test)] mod test { - use std::time::UNIX_EPOCH; + use std::time::{Duration, UNIX_EPOCH}; use catalog::local::MemoryCatalogManager; use catalog::{CatalogManager, RegisterTableRequest}; @@ -674,7 +705,10 @@ mod test { use datatypes::prelude::ConcreteDataType; use datatypes::schema::{ColumnSchema, Schema}; use promql_parser::label::Matcher; - use promql_parser::parser::ValueType; + use promql_parser::parser::{ + BinModifier, FunctionArgs as PromFunctionArgs, ValueType, VectorMatchCardinality, + VectorMatchModifier, + }; use query::query_engine::QueryEngineState; use query::DfContextProviderAdapter; use session::context::QueryContext; @@ -765,25 +799,29 @@ mod test { variadic: false, return_type: ValueType::Vector, }, - args: vec![Box::new(PromExpr::VectorSelector(VectorSelector { - name: Some("some_metric".to_owned()), - offset: None, - start_or_end: None, - label_matchers: Matchers { - matchers: vec![ - Matcher { - op: MatchOp::NotEqual, - name: "tag_0".to_string(), - value: "bar".to_string(), - }, - Matcher { - op: MatchOp::Equal, - name: METRIC_NAME.to_string(), - value: "some_metric".to_string(), - }, - ], - }, - }))], + args: PromFunctionArgs { + args: vec![Box::new(PromExpr::VectorSelector(VectorSelector { + name: Some("some_metric".to_owned()), + offset: None, + at: None, + label_matchers: Matchers { + matchers: vec![ + Matcher { + op: MatchOp::NotEqual, + name: "tag_0".to_string(), + value: "bar".to_string(), + }, + Matcher { + op: MatchOp::Equal, + name: METRIC_NAME.to_string(), + value: "some_metric".to_string(), + }, + ] + .into_iter() + .collect(), + }, + }))], + }, }); let eval_stmt = EvalStmt { expr: prom_expr, @@ -986,7 +1024,7 @@ mod test { expr: Box::new(PromExpr::VectorSelector(VectorSelector { name: Some("some_metric".to_owned()), offset: None, - start_or_end: None, + at: None, label_matchers: Matchers { matchers: vec![ Matcher { @@ -999,12 +1037,13 @@ mod test { name: METRIC_NAME.to_string(), value: "some_metric".to_string(), }, - ], + ] + .into_iter() + .collect(), }, })), - param: Box::new(PromExpr::empty_vector_selector()), - grouping: vec![String::from("tag_1")], - without: false, + param: None, + grouping: AggModifier::By(vec![String::from("tag_1")].into_iter().collect()), }); let mut eval_stmt = EvalStmt { expr: prom_expr, @@ -1032,8 +1071,8 @@ mod test { ); // test group without - if let PromExpr::Aggregate(AggregateExpr { without, .. }) = &mut eval_stmt.expr { - *without = true; + if let PromExpr::Aggregate(AggregateExpr { grouping, .. }) = &mut eval_stmt.expr { + *grouping = AggModifier::Without(vec![String::from("tag_1")].into_iter().collect()); } let context_provider = build_test_context_provider("some_metric".to_string(), 2, 2).await; let plan = PromPlanner::stmt_to_plan(eval_stmt, context_provider).unwrap(); @@ -1143,7 +1182,7 @@ mod test { lhs: Box::new(PromExpr::VectorSelector(VectorSelector { name: Some("some_metric".to_owned()), offset: None, - start_or_end: None, + at: None, label_matchers: Matchers { matchers: vec![ Matcher { @@ -1156,14 +1195,16 @@ mod test { name: METRIC_NAME.to_string(), value: "some_metric".to_string(), }, - ], + ] + .into_iter() + .collect(), }, })), op: token::T_ADD, rhs: Box::new(PromExpr::VectorSelector(VectorSelector { name: Some("some_metric".to_owned()), offset: None, - start_or_end: None, + at: None, label_matchers: Matchers { matchers: vec![ Matcher { @@ -1176,11 +1217,16 @@ mod test { name: METRIC_NAME.to_string(), value: "some_metric".to_string(), }, - ], + ] + .into_iter() + .collect(), }, })), - matching: None, - return_bool: false, + matching: BinModifier { + card: VectorMatchCardinality::OneToOne, + matching: VectorMatchModifier::Ignoring(HashSet::new()), + return_bool: false, + }, }); let eval_stmt = EvalStmt { @@ -1221,7 +1267,7 @@ mod test { rhs: Box::new(PromExpr::VectorSelector(VectorSelector { name: Some("some_metric".to_owned()), offset: None, - start_or_end: None, + at: None, label_matchers: Matchers { matchers: vec![ Matcher { @@ -1234,11 +1280,16 @@ mod test { name: METRIC_NAME.to_string(), value: "some_metric".to_string(), }, - ], + ] + .into_iter() + .collect(), }, })), - matching: None, - return_bool: false, + matching: BinModifier { + card: VectorMatchCardinality::OneToOne, + matching: VectorMatchModifier::Ignoring(HashSet::new()), + return_bool: false, + }, }); let eval_stmt = EvalStmt { @@ -1271,8 +1322,11 @@ mod test { lhs: Box::new(PromExpr::NumberLiteral(NumberLiteral { val: 1.0 })), op: token::T_ADD, rhs: Box::new(PromExpr::NumberLiteral(NumberLiteral { val: 1.0 })), - matching: None, - return_bool: false, + matching: BinModifier { + card: VectorMatchCardinality::OneToOne, + matching: VectorMatchModifier::Ignoring(HashSet::new()), + return_bool: false, + }, }); let eval_stmt = EvalStmt { diff --git a/src/query/Cargo.toml b/src/query/Cargo.toml index 45edb79e2fd..2ba78232d46 100644 --- a/src/query/Cargo.toml +++ b/src/query/Cargo.toml @@ -27,7 +27,7 @@ futures-util.workspace = true metrics = "0.20" once_cell = "1.10" promql = { path = "../promql" } -promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "fec3c8bcee982b8add2a77d07818d3cbe92b89fe" } +promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "d027ce428a6a2df5a652b8558608c77d33c31644" } serde.workspace = true serde_json = "1.0" session = { path = "../session" } diff --git a/src/query/src/parser.rs b/src/query/src/parser.rs index f8b696c238f..2e697aa9f6a 100644 --- a/src/query/src/parser.rs +++ b/src/query/src/parser.rs @@ -14,10 +14,11 @@ use std::time::Duration; +use common_error::ext::PlainError; use common_error::prelude::BoxedError; +use common_error::status_code::StatusCode; use common_telemetry::timer; -use promql_parser::label::{MatchOp, Matcher, Matchers}; -use promql_parser::parser::{EvalStmt, Expr as PromExpr, Function, ValueType}; +use promql_parser::parser::EvalStmt; use snafu::ResultExt; use sql::dialect::GenericDialect; use sql::parser::ParserContext; @@ -53,38 +54,13 @@ impl QueryLanguageParser { } // TODO(ruihang): implement this method when parser is ready. - pub fn parse_promql(_promql: &str) -> Result { + pub fn parse_promql(promql: &str) -> Result { let _timer = timer!(METRIC_PARSE_PROMQL_ELAPSED); - let prom_expr = PromExpr::Call(promql_parser::parser::Call { - func: Function { - name: "ceil", - arg_types: vec![ValueType::Vector], - variadic: false, - return_type: ValueType::Vector, - }, - args: vec![Box::new(PromExpr::VectorSelector( - promql_parser::parser::VectorSelector { - name: Some("demo".to_owned()), - offset: None, - start_or_end: None, - label_matchers: Matchers { - matchers: vec![ - Matcher { - op: MatchOp::Equal, - name: "host".to_string(), - value: "host1".to_string(), - }, - Matcher { - op: MatchOp::Equal, - name: promql_parser::label::METRIC_NAME.to_string(), - value: "demo".to_string(), - }, - ], - }, - }, - ))], - }); + let prom_expr = promql_parser::parser::parse(promql) + .map_err(|msg| BoxedError::new(PlainError::new(msg, StatusCode::InvalidArguments))) + .context(QueryParseSnafu { query: promql })?; + let eval_stmt = EvalStmt { expr: prom_expr, start: std::time::UNIX_EPOCH,