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_with_alias( table_provider: DfTableSourceProvider, stmt: &EvalStmt, alias: Option<String>, query_engine_state: &QueryEngineState, ) -> Result<LogicalPlan>

pub async fn prom_expr_to_plan( &mut self, prom_expr: &PromExpr, query_engine_state: &QueryEngineState, ) -> Result<LogicalPlan>

fn prom_expr_to_plan_inner<'life0, 'life1, 'life_self, 'async_recursion>( &'life_self mut self, prom_expr: &'life0 PromExpr, timestamp_fn: bool, query_engine_state: &'life1 QueryEngineState, ) -> Pin<Box<dyn Future<Output = Result<LogicalPlan>> + Send + 'async_recursion>>

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

Converts a PromQL expression to a logical plan.

NOTE: The timestamp_fn indicates whether the PromQL timestamp() function is being evaluated in the current context. If true, the planner generates a logical plan that projects the timestamp (time index) column as the value column for each input row, implementing the PromQL timestamp() function semantics. If false, the planner generates the standard logical plan for the given PromQL expression.

async fn prom_subquery_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, subquery_expr: &SubqueryExpr, ) -> Result<LogicalPlan>

async fn prom_aggr_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, aggr_expr: &AggregateExpr, ) -> Result<LogicalPlan>

async fn prom_topk_bottomk_to_plan( &mut self, aggr_expr: &AggregateExpr, input: LogicalPlan, ) -> Result<LogicalPlan>

Create logical plan for PromQL topk and bottomk expr.

async fn prom_unary_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, unary_expr: &UnaryExpr, ) -> Result<LogicalPlan>

async fn prom_binary_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, binary_expr: &PromBinaryExpr, ) -> Result<LogicalPlan>

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

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

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

fn create_timestamp_func_plan( &mut self, normalize: LogicalPlan, ) -> Result<LogicalPlan>

Builds a projection plan for the PromQL timestamp() function. Projects the time index column as the value column for each row.

Arguments

• normalize - Input [LogicalPlan] for the normalized series.

Returns

Returns a Result<LogicalPlan> where the resulting logical plan projects the timestamp column as the value column, along with the original tag and time index columns.

Timestamp vs. Time Function

• Timestamp Function (timestamp()): In PromQL, the timestamp() function returns the timestamp (time index) of each sample as the value column.

• Time Function (time()): The time() function returns the evaluation time of the query as a scalar value.

Side Effects

Updates the planner context’s field columns to the timestamp column name.

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

async fn prom_call_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, call_expr: &Call, ) -> Result<LogicalPlan>

async fn prom_ext_expr_to_plan( &mut self, query_engine_state: &QueryEngineState, ext_expr: &Extension, ) -> Result<LogicalPlan>

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

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>

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

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 if update_ctx is true.

pub fn matchers_to_expr( label_matchers: Matchers, table_schema: &DFSchemaRef, ) -> Result<Vec<DfExpr>>

fn find_case_sensitive_column( schema: &DFSchemaRef, column: &str, ) -> Option<String>

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

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

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

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<Option<LogicalPlan>>

Setup PromPlannerContext’s state fields.

Returns a logical plan for an empty metric.

fn setup_context_for_empty_metric(&mut self) -> Result<LogicalPlan>

Setup PromPlannerContext’s state fields for a non existent table without any rows.

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

fn create_function_expr( &mut self, func: &Function, other_input_exprs: Vec<DfExpr>, query_engine_state: &QueryEngineState, ) -> Result<(Vec<DfExpr>, Vec<String>)>

Creates function expressions for projection and returns the expressions and new tags.

Side Effects

This method will update PromPlannerContext’s fields and tags if needed.

fn validate_label_name(label_name: &str) -> Result<()>

Validate label name according to Prometheus specification. Label names must match the regex: [a-zA-Z_][a-zA-Z0-9_]* Additionally, label names starting with double underscores are reserved for internal use.

fn build_regexp_replace_label_expr( &self, other_input_exprs: &mut VecDeque<DfExpr>, query_engine_state: &QueryEngineState, ) -> Result<Option<(DfExpr, String)>>

Build expr for label_replace function

fn build_concat_labels_expr( other_input_exprs: &mut VecDeque<DfExpr>, ctx: &PromPlannerContext, query_engine_state: &QueryEngineState, ) -> Result<(DfExpr, String)>

Build expr for label_join function

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

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

fn create_field_column_exprs(&self) -> Result<Vec<DfExpr>>

fn create_tag_and_time_index_column_sort_exprs(&self) -> Result<Vec<SortExpr>>

fn create_field_columns_sort_exprs(&self, asc: bool) -> Vec<SortExpr>

fn create_sort_exprs_by_tags( func: &str, tags: Vec<DfExpr>, asc: bool, ) -> Result<Vec<SortExpr>>

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

fn create_aggregate_exprs( &mut self, op: TokenType, param: &Option<Box<PromExpr>>, input_plan: &LogicalPlan, ) -> Result<(Vec<DfExpr>, Vec<DfExpr>)>

Creates a set of DataFusion DfExpr::AggregateFunction expressions for each value column using the specified aggregate function.

Side Effects

This method modifies the value columns in the context by replacing them with the new columns created by the aggregate function application.

Returns

Returns a tuple of (aggregate_expressions, previous_field_expressions) where:

• aggregate_expressions: Expressions that apply the aggregate function to the original fields
• previous_field_expressions: Original field expressions before aggregation. This is non-empty only when the operation is count_values, as this operation requires preserving the original values for grouping.

fn get_param_value_as_str( op: TokenType, param: &Option<Box<PromExpr>>, ) -> Result<&str>

fn get_param_as_literal_expr( param: &Option<Box<PromExpr>>, op: Option<TokenType>, expected_type: Option<ArrowDataType>, ) -> Result<DfExpr>

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

Create [DfExpr::WindowFunction] expr for each value column with given window function.

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

👎Deprecated: use Self::get_param_as_literal_expr instead. This is only for create_histogram_plan

Try to build a f64 from [PromExpr].

async fn create_histogram_plan( &mut self, args: &PromFunctionArgs, query_engine_state: &QueryEngineState, ) -> Result<LogicalPlan>

Create a SPECIAL_HISTOGRAM_QUANTILE plan.

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

Create a SPECIAL_VECTOR_FUNCTION plan

async fn create_scalar_plan( &mut self, args: &PromFunctionArgs, query_engine_state: &QueryEngineState, ) -> Result<LogicalPlan>

Create a SCALAR_FUNCTION plan

async fn create_absent_plan( &mut self, args: &PromFunctionArgs, query_engine_state: &QueryEngineState, ) -> Result<LogicalPlan>

Create a SPECIAL_ABSENT_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_with_context( &self, expr: &PromExpr, ) -> Option<DfExpr>

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

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, left_time_index_column: Option<String>, right_time_index_column: Option<String>, only_join_time_index: bool, modifier: &Option<BinModifier>, ) -> Result<LogicalPlan>

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>

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>

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

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

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>

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

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>

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

fn apply_alias_projection( &mut self, plan: LogicalPlan, alias_name: String, ) -> Result<LogicalPlan>

Apply an alias to the query result by adding a projection with the alias name