Struct PromPlanner

pub struct PromPlanner {
    table_provider: DfTableSourceProvider,
    ctx: PromPlannerContext,
}

Fields

table_provider: DfTableSourceProvider

ctx: PromPlannerContext

Implementations

impl PromPlanner

pub async fn stmt_to_plan( table_provider: DfTableSourceProvider, stmt: &EvalStmt, session_state: &SessionState, ) -> Result<LogicalPlan, Error>

pub fn prom_expr_to_plan<'life0, 'life1, 'life_self, 'async_recursion>( &'life_self mut self, prom_expr: &'life0 PromExpr, session_state: &'life1 SessionState, ) -> Pin<Box<dyn Future<Output = Result<LogicalPlan, Error>> + Send + 'async_recursion>>

where 'life0: 'async_recursion, 'life1: 'async_recursion, 'life_self: 'async_recursion,

async fn prom_aggr_expr_to_plan( &mut self, session_state: &SessionState, aggr_expr: &AggregateExpr, ) -> Result<LogicalPlan, Error>

async fn prom_unary_expr_to_plan( &mut self, session_state: &SessionState, unary_expr: &UnaryExpr, ) -> Result<LogicalPlan, Error>

async fn prom_binary_expr_to_plan( &mut self, session_state: &SessionState, binary_expr: &PromBinaryExpr, ) -> Result<LogicalPlan, Error>

fn prom_number_lit_to_plan( &mut self, number_literal: &NumberLiteral, ) -> Result<LogicalPlan, Error>

fn prom_string_lit_to_plan( &mut self, string_literal: &StringLiteral, ) -> Result<LogicalPlan, Error>

async fn prom_vector_selector_to_plan( &mut self, vector_selector: &VectorSelector, ) -> Result<LogicalPlan, Error>

async fn prom_matrix_selector_to_plan( &mut self, matrix_selector: &MatrixSelector, ) -> Result<LogicalPlan, Error>

async fn prom_call_expr_to_plan( &mut self, session_state: &SessionState, call_expr: &Call, ) -> Result<LogicalPlan, Error>

async fn prom_ext_expr_to_plan( &mut self, session_state: &SessionState, ext_expr: &Extension, ) -> Result<LogicalPlan, Error>

fn preprocess_label_matchers( &mut self, label_matchers: &Matchers, name: &Option<String>, ) -> Result<Matchers, Error>

Extract metric name from __name__ matcher and set it into PromPlannerContext. Returns a new [Matchers] that doesn’t contain metric name matcher.

Each call to this function means new selector is started. Thus, the context will be reset at first.

Name rule:

• if name is some, then the matchers MUST NOT contain __name__ matcher.
• if name is none, then the matchers MAY contain NONE OR MULTIPLE __name__ matchers.

async fn selector_to_series_normalize_plan( &mut self, offset: &Option<Offset>, label_matchers: Matchers, is_range_selector: bool, ) -> Result<LogicalPlan, Error>

fn agg_modifier_to_col( &mut self, input_schema: &DFSchemaRef, modifier: &Option<LabelModifier>, ) -> Result<Vec<DfExpr>, Error>

Convert [LabelModifier] to [Column] exprs for aggregation. Timestamp column and tag columns will be included.

Side effect

This method will also change the tag columns in ctx.

fn matchers_to_expr( &self, label_matchers: Matchers, ) -> Result<Vec<DfExpr>, Error>

fn table_ref(&self) -> Result<TableReference, Error>

fn build_time_index_filter( &self, offset_duration: i64, ) -> Result<Option<DfExpr>, Error>

async fn create_table_scan_plan( &mut self, table_ref: TableReference, filter: Vec<DfExpr>, ) -> Result<LogicalPlan, Error>

Create a table scan plan and a filter plan with given filter.

Panic

If the filter is empty

async fn setup_context(&mut self) -> Result<(), Error>

Setup PromPlannerContext’s state fields.

fn create_function_args( &self, args: &[Box<PromExpr>], ) -> Result<FunctionArgs, Error>

fn create_function_expr( &mut self, func: &Function, other_input_exprs: Vec<DfExpr>, session_state: &SessionState, ) -> Result<Vec<DfExpr>, Error>

Side Effects

This method will update PromPlannerContext’s value fields.

fn build_regexp_replace_label_expr( other_input_exprs: &mut VecDeque<DfExpr>, session_state: &SessionState, ) -> Result<(DfExpr, String), Error>

Build expr for label_replace function

fn build_concat_labels_expr( other_input_exprs: &mut VecDeque<DfExpr>, session_state: &SessionState, ) -> Result<(DfExpr, String), Error>

Build expr for label_join function

fn create_time_index_column_expr(&self) -> Result<DfExpr, Error>

fn create_tag_column_exprs(&self) -> Result<Vec<DfExpr>, Error>

fn create_tag_and_time_index_column_sort_exprs( &self, ) -> Result<Vec<DfExpr>, Error>

fn create_empty_values_filter_expr(&self) -> Result<DfExpr, Error>

fn create_aggregate_exprs( &mut self, op: TokenType, input_plan: &LogicalPlan, ) -> Result<Vec<DfExpr>, Error>

Create [DfExpr::AggregateFunction] expr for each value column with given aggregate function.

Side effect

This method will update value columns in context to the new value columns created by aggregate function.

async fn create_histogram_plan( &mut self, args: &PromFunctionArgs, session_state: &SessionState, ) -> Result<LogicalPlan, Error>

Create a SPECIAL_HISTOGRAM_QUANTILE plan.

async fn create_vector_plan( &mut self, args: &PromFunctionArgs, ) -> Result<LogicalPlan, Error>

Create a SPECIAL_VECTOR_FUNCTION plan

async fn create_scalar_plan( &mut self, args: &PromFunctionArgs, session_state: &SessionState, ) -> Result<LogicalPlan, Error>

Create a SCALAR_FUNCTION plan

fn try_build_literal_expr(expr: &PromExpr) -> Option<DfExpr>

Try to build a DataFusion Literal Expression from PromQL Expr, return None if the input is not a literal expression.

fn try_build_special_time_expr( expr: &PromExpr, time_index_col: &str, ) -> Option<DfExpr>

fn try_build_float_literal(expr: &PromExpr) -> Option<f64>

Try to build a f64 from [PromExpr].

fn prom_token_to_binary_expr_builder( token: TokenType, ) -> Result<Box<dyn Fn(DfExpr, DfExpr) -> Result<DfExpr, Error>>, Error>

Return a lambda to build binary expression from token. Because some binary operator are function in DataFusion like atan2 or ^.

fn is_token_a_comparison_op(token: TokenType) -> bool

Check if the given op is a comparison operator.

fn is_token_a_set_op(token: TokenType) -> bool

Check if the given op is a set operator (UNION, INTERSECT and EXCEPT in SQL).

fn join_on_non_field_columns( &self, left: LogicalPlan, right: LogicalPlan, left_table_ref: TableReference, right_table_ref: TableReference, only_join_time_index: bool, ) -> Result<LogicalPlan, Error>

Build a inner join on time index column and tag columns to concat two logical plans. When only_join_time_index == true we only join on the time index, because these two plan may not have the same tag columns

fn set_op_on_non_field_columns( &mut self, left: LogicalPlan, right: LogicalPlan, left_context: PromPlannerContext, right_context: PromPlannerContext, op: TokenType, modifier: &Option<BinModifier>, ) -> Result<LogicalPlan, Error>

Build a set operator (AND/OR/UNLESS)

fn or_operator( &mut self, left: LogicalPlan, right: LogicalPlan, left_tag_cols_set: HashSet<String>, right_tag_cols_set: HashSet<String>, left_context: PromPlannerContext, right_context: PromPlannerContext, modifier: &Option<BinModifier>, ) -> Result<LogicalPlan, Error>

fn projection_for_each_field_column<F>( &mut self, input: LogicalPlan, name_to_expr: F, ) -> Result<LogicalPlan, Error>

where F: FnMut(&String) -> Result<DfExpr, Error>,

Build a projection that project and perform operation expr for every value columns. Non-value columns (tag and timestamp) will be preserved in the projection.

Side effect

This function will update the value columns in the context. Those new column names don’t contains qualifier.

fn filter_on_field_column<F>( &self, input: LogicalPlan, name_to_expr: F, ) -> Result<LogicalPlan, Error>

where F: FnMut(&String) -> Result<DfExpr, Error>,

Build a filter plan that filter on value column. Notice that only one value column is expected.

fn date_part_on_time_index(&self, date_part: &str) -> Result<DfExpr, Error>

Generate an expr like date_part("hour", <TIME_INDEX>). Caller should ensure the time index column in context is set