Struct LogicalAgg

pub struct LogicalAgg {
    pub base: PlanBase<Logical>,
    core: Agg<PlanRef>,
}

LogicalAgg groups input data by their group key and computes aggregation functions.

It corresponds to the GROUP BY operator in a SQL query statement together with the aggregate functions in the SELECT clause.

The output schema will first include the group key and then the aggregation calls.

Fields

base: PlanBase<Logical>

core: Agg<PlanRef>

Implementations

impl LogicalAgg

fn gen_stateless_two_phase_streaming_agg_plan( &self, stream_input: PlanRef, ) -> Result<PlanRef, RwError>

Generate plan for stateless 2-phase streaming agg. Should only be used iff input is distributed. Input must be converted to stream form.

fn gen_vnode_two_phase_streaming_agg_plan( &self, stream_input: PlanRef, dist_key: &[usize], ) -> Result<PlanRef, RwError>

Generate plan for stateless/stateful 2-phase streaming agg. Should only be used iff input is distributed. Input must be converted to stream form.

fn gen_single_plan(&self, stream_input: PlanRef) -> Result<PlanRef, RwError>

fn gen_shuffle_plan(&self, stream_input: PlanRef) -> Result<PlanRef, RwError>

fn gen_dist_stream_agg_plan( &self, stream_input: PlanRef, ) -> Result<PlanRef, RwError>

Generates distributed stream plan.

fn prepare_approx_percentile( &self, core: &mut Agg<PlanRef>, stream_input: PlanRef, ) -> Result<(ColIndexMapping, ColIndexMapping, Option<PlanRef>), RwError>

Prepares metadata and the approx_percentile plan, if there’s one present. It may modify core.agg_calls to separate normal agg and approx percentile agg, and core.input to share the input via StreamShare, to both approx percentile agg and normal agg.

fn need_row_merge(approx_percentile: &Option<PlanRef>) -> bool

fn add_row_merge_if_needed( approx_percentile: Option<PlanRef>, global_agg: PlanRef, approx_percentile_col_mapping: ColIndexMapping, non_approx_percentile_col_mapping: ColIndexMapping, ) -> Result<PlanRef, RwError>

Add RowMerge if needed

fn separate_normal_and_special_agg(&self) -> SeparatedAggInfo

fn build_approx_percentile_agg( &self, input: PlanRef, approx_percentile_agg_call: &PlanAggCall, ) -> Result<PlanRef, RwError>

fn build_approx_percentile_aggs( &self, input: PlanRef, approx_percentile_agg_call: &[PlanAggCall], ) -> Result<Option<PlanRef>, RwError>

If only 1 approx percentile, just return it. Otherwise build a tree of approx percentile with MergeProject. e.g. ApproxPercentile(col1, 0.5) as x, ApproxPercentile(col2, 0.5) as y, ApproxPercentile(col3, 0.5) as z will be built as MergeProject /
MergeProject z /
x y

pub fn core(&self) -> &Agg<PlanRef>

impl LogicalAgg

pub fn i2o_col_mapping(&self) -> ColIndexMapping

get the Mapping of columnIndex from input column index to out column index

pub fn create( select_exprs: Vec<ExprImpl>, group_by: GroupBy, having: Option<ExprImpl>, input: PlanRef, ) -> Result<(PlanRef, Vec<ExprImpl>, Option<ExprImpl>), RwError>

create will analyze select exprs, group exprs and having, and construct a plan like

LogicalAgg -> LogicalProject -> input

It also returns the rewritten select exprs and having that reference into the aggregated results.

pub fn agg_calls(&self) -> &Vec<PlanAggCall>

Get a reference to the logical agg’s agg calls.

pub fn group_key(&self) -> &IndexSet

Get a reference to the logical agg’s group key.

pub fn grouping_sets(&self) -> &Vec<IndexSet>

pub fn decompose( self, ) -> (Vec<PlanAggCall>, IndexSet, Vec<IndexSet>, PlanRef, bool)

pub fn rewrite_with_input_agg( &self, input: PlanRef, agg_calls: &[PlanAggCall], input_col_change: ColIndexMapping, ) -> (Self, ColIndexMapping)

Methods from Deref<Target = PlanBase<Logical>>

pub fn clone_with_new_plan_id(&self) -> Self

pub fn clone_with_new_distribution(&self, dist: Distribution) -> Self

Clone the plan node with a new distribution.

Panics if the plan node is not physical.

Trait Implementations

impl Clone for LogicalAgg

fn clone(&self) -> LogicalAgg

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl ColPrunable for LogicalAgg

fn prune_col( &self, required_cols: &[usize], ctx: &mut ColumnPruningContext, ) -> PlanRef

Transform the plan node to only output the required columns ordered by index number.

impl Debug for LogicalAgg

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Deref for LogicalAgg

type Target = PlanBase<Logical>

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl Distill for LogicalAgg

fn distill<'a>(&self) -> XmlNode<'a>

fn distill_to_string(&self) -> String

impl ExprRewritable for LogicalAgg

fn has_rewritable_expr(&self) -> bool

fn rewrite_exprs(&self, r: &mut dyn ExprRewriter) -> PlanRef

impl ExprVisitable for LogicalAgg

fn visit_exprs(&self, v: &mut dyn ExprVisitor)

impl From<Agg<PlanRef>> for LogicalAgg

fn from(core: Agg<PlanRef>) -> Self

Converts to this type from the input type.

impl Hash for LogicalAgg

fn hash<__H: Hasher>(&self, state: &mut __H)

Feeds this value into the given Hasher.

fn hash_slice<H>(data: &[Self], state: &mut H)

where H: Hasher, Self: Sized,

Feeds a slice of this type into the given Hasher.

impl PartialEq for LogicalAgg

fn eq(&self, other: &LogicalAgg) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PlanNodeMeta for LogicalAgg

const NODE_TYPE: PlanNodeType = PlanNodeType::LogicalAgg

type Convention = Logical

fn plan_base(&self) -> &PlanBase<Logical>

Get the reference to the PlanBase with corresponding convention.

fn plan_base_ref(&self) -> PlanBaseRef<'_>

Get the reference to the PlanBase with erased convention.

impl PlanTreeNode for LogicalAgg

fn clone_with_inputs(&self, inputs: &[PlanRef]) -> PlanRef

Clone the node with a list of new inputs.

fn inputs(&self) -> SmallVec<[PlanRef; 2]>

Get input nodes of the plan.

impl PlanTreeNodeUnary for LogicalAgg

fn input(&self) -> PlanRef

fn clone_with_input(&self, input: PlanRef) -> Self

fn rewrite_with_input( &self, input: PlanRef, input_col_change: ColIndexMapping, ) -> (Self, ColIndexMapping)

Rewrites the plan node according to the schema change of its input node during rewriting.

impl PredicatePushdown for LogicalAgg

fn predicate_pushdown( &self, predicate: Condition, ctx: &mut PredicatePushdownContext, ) -> PlanRef

Push predicate down for every logical plan node.

impl ToBatch for LogicalAgg

fn to_batch(&self) -> Result<PlanRef, RwError>

to_batch is equivalent to to_batch_with_order_required(&Order::any())

fn to_batch_with_order_required( &self, required_order: &Order, ) -> Result<PlanRef, RwError>

convert the plan to batch physical plan and satisfy the required Order

impl ToDistributedBatch for LogicalAgg

fn to_distributed(&self) -> Result<PlanRef, RwError>

to_distributed is equivalent to to_distributed_with_required(&Order::any(), &RequiredDist::Any)

fn to_distributed_with_required( &self, required_order: &Order, required_dist: &RequiredDist, ) -> Result<PlanRef, RwError>

insert the exchange in batch physical plan to satisfy the required Distribution and Order.

impl ToLocalBatch for LogicalAgg

fn to_local(&self) -> Result<PlanRef, RwError>

fn to_local_with_order_required( &self, required_order: &Order, ) -> Result<PlanRef, RwError>

Convert the plan to batch local physical plan and satisfy the required Order

impl ToStream for LogicalAgg

fn to_stream(&self, ctx: &mut ToStreamContext) -> Result<PlanRef, RwError>

to_stream is equivalent to to_stream_with_dist_required(RequiredDist::Any)

fn logical_rewrite_for_stream( &self, ctx: &mut RewriteStreamContext, ) -> Result<(PlanRef, ColIndexMapping), RwError>

logical_rewrite_for_stream will rewrite the logical node, and return (new_plan_node, col_mapping), the col_mapping is for original columns have been changed into some other position.

fn to_stream_with_dist_required( &self, required_dist: &RequiredDist, ctx: &mut ToStreamContext, ) -> Result<PlanRef, RwError>

convert the plan to streaming physical plan and satisfy the required distribution

impl TryToBatchPb for LogicalAgg

fn try_to_batch_prost_body(&self) -> Result<NodeBody, SchedulerError>

impl TryToStreamPb for LogicalAgg

fn try_to_stream_prost_body( &self, _state: &mut BuildFragmentGraphState, ) -> Result<NodeBody, SchedulerError>

impl Eq for LogicalAgg

impl PlanNode for LogicalAgg

impl StructuralPartialEq for LogicalAgg

impl ToPb for LogicalAgg