From 24222de93ae76df88eb7dffe8ea63ea60f60e13d Mon Sep 17 00:00:00 2001 From: ice1000 Date: Sun, 4 Jun 2023 01:04:28 +0000 Subject: [PATCH 01/22] Save --- .../src/optimizer/plan_node/generic/agg.rs | 28 +++++++++++++------ 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/agg.rs b/src/frontend/src/optimizer/plan_node/generic/agg.rs index 02e2985a0c61..bfbd310b3fc0 100644 --- a/src/frontend/src/optimizer/plan_node/generic/agg.rs +++ b/src/frontend/src/optimizer/plan_node/generic/agg.rs @@ -17,6 +17,7 @@ use std::fmt; use fixedbitset::FixedBitSet; use itertools::{Either, Itertools}; +use pretty_xmlish::Pretty; use risingwave_common::catalog::{Field, FieldDisplay, Schema}; use risingwave_common::types::DataType; use risingwave_common::util::sort_util::{ColumnOrder, ColumnOrderDisplay, OrderType}; @@ -597,20 +598,31 @@ impl Agg { } fn agg_calls_display(&self) -> Vec> { - self.agg_calls - .iter() - .map(|plan_agg_call| PlanAggCallDisplay { + let f = |plan_agg_call| PlanAggCallDisplay { + plan_agg_call, + input_schema: self.input.schema(), + }; + self.agg_calls.iter().map(f).collect_vec() + } + + fn agg_calls_pretty(&self) -> Pretty<'_> { + let f = |plan_agg_call| { + Pretty::debug(&PlanAggCallDisplay { plan_agg_call, input_schema: self.input.schema(), }) - .collect_vec() + }; + Pretty::Array(self.agg_calls.iter().map(f).collect()) } fn group_key_display(&self) -> Vec> { - self.group_key - .ones() - .map(|i| FieldDisplay(self.input.schema().fields.get(i).unwrap())) - .collect_vec() + let f = |i| FieldDisplay(self.input.schema().fields.get(i).unwrap()); + self.group_key.ones().map(f).collect_vec() + } + + fn group_key_pretty(&self) -> Pretty<'_> { + let f = |i| Pretty::display(&FieldDisplay(self.input.schema().fields.get(i).unwrap())); + Pretty::Array(self.group_key.ones().map(f).collect()) } } From 46c789638999e9c1217e427a20ede25d334287e8 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Sun, 4 Jun 2023 01:44:54 +0000 Subject: [PATCH 02/22] Mua --- .../src/optimizer/plan_node/generic/agg.rs | 16 +++++++++++++--- .../plan_node/generic/dynamic_filter.rs | 10 +++++++--- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/agg.rs b/src/frontend/src/optimizer/plan_node/generic/agg.rs index bfbd310b3fc0..e13cf98ef6a7 100644 --- a/src/frontend/src/optimizer/plan_node/generic/agg.rs +++ b/src/frontend/src/optimizer/plan_node/generic/agg.rs @@ -597,15 +597,25 @@ impl Agg { builder.field("aggs", &self.agg_calls_display()); } + pub fn fields_pretty<'a>(&self) -> Vec<(&'a str, Pretty<'a>)> { + let last = ("aggs", self.agg_calls_pretty()); + if self.group_key.count_ones(..) != 0 { + let first = ("group_key", self.group_key_pretty()); + vec![first, last] + } else { + vec![last] + } + } + fn agg_calls_display(&self) -> Vec> { let f = |plan_agg_call| PlanAggCallDisplay { plan_agg_call, input_schema: self.input.schema(), }; - self.agg_calls.iter().map(f).collect_vec() + self.agg_calls.iter().map(f).collect() } - fn agg_calls_pretty(&self) -> Pretty<'_> { + fn agg_calls_pretty<'a>(&self) -> Pretty<'a> { let f = |plan_agg_call| { Pretty::debug(&PlanAggCallDisplay { plan_agg_call, @@ -620,7 +630,7 @@ impl Agg { self.group_key.ones().map(f).collect_vec() } - fn group_key_pretty(&self) -> Pretty<'_> { + fn group_key_pretty<'a>(&self) -> Pretty<'a> { let f = |i| Pretty::display(&FieldDisplay(self.input.schema().fields.get(i).unwrap())); Pretty::Array(self.group_key.ones().map(f).collect()) } diff --git a/src/frontend/src/optimizer/plan_node/generic/dynamic_filter.rs b/src/frontend/src/optimizer/plan_node/generic/dynamic_filter.rs index f4560b0cb29d..727d4d6f04ad 100644 --- a/src/frontend/src/optimizer/plan_node/generic/dynamic_filter.rs +++ b/src/frontend/src/optimizer/plan_node/generic/dynamic_filter.rs @@ -122,18 +122,22 @@ impl DynamicFilter { watermark_columns } - pub fn fmt_fields_with_builder(&self, builder: &mut fmt::DebugStruct<'_, '_>) { + fn condition_display(&self) -> (Condition, Schema) { let mut concat_schema = self.left.schema().fields.clone(); concat_schema.extend(self.right.schema().fields.clone()); let concat_schema = Schema::new(concat_schema); let predicate = self.predicate(); + (predicate, concat_schema) + } + pub fn fmt_fields_with_builder(&self, builder: &mut fmt::DebugStruct<'_, '_>) { + let (condition, input_schema) = &self.condition_display(); builder.field( "predicate", &ConditionDisplay { - condition: &predicate, - input_schema: &concat_schema, + condition, + input_schema, }, ); } From b0943c1a3cfc6118237a5d3cb56f78e233aa4b5f Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 03:50:47 +0000 Subject: [PATCH 03/22] Further prepare --- .../src/optimizer/plan_node/generic/agg.rs | 8 ++-- .../optimizer/plan_node/generic/hop_window.rs | 47 ++++++++++++++++++ .../optimizer/plan_node/generic/intersect.rs | 7 +++ .../src/optimizer/plan_node/generic/mod.rs | 12 +++++ .../optimizer/plan_node/generic/project.rs | 48 +++++++++++-------- .../src/optimizer/plan_node/generic/union.rs | 6 +++ 6 files changed, 104 insertions(+), 24 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/agg.rs b/src/frontend/src/optimizer/plan_node/generic/agg.rs index e13cf98ef6a7..be800a708995 100644 --- a/src/frontend/src/optimizer/plan_node/generic/agg.rs +++ b/src/frontend/src/optimizer/plan_node/generic/agg.rs @@ -17,7 +17,7 @@ use std::fmt; use fixedbitset::FixedBitSet; use itertools::{Either, Itertools}; -use pretty_xmlish::Pretty; +use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::{Field, FieldDisplay, Schema}; use risingwave_common::types::DataType; use risingwave_common::util::sort_util::{ColumnOrder, ColumnOrderDisplay, OrderType}; @@ -35,7 +35,7 @@ use crate::stream_fragmenter::BuildFragmentGraphState; use crate::utils::{ ColIndexMapping, ColIndexMappingRewriteExt, Condition, ConditionDisplay, IndexRewriter, }; -use crate::TableCatalog; +use crate::{TableCatalog, impl_distill_unit_from_fields}; /// [`Agg`] groups input data by their group key and computes aggregation functions. /// @@ -597,7 +597,7 @@ impl Agg { builder.field("aggs", &self.agg_calls_display()); } - pub fn fields_pretty<'a>(&self) -> Vec<(&'a str, Pretty<'a>)> { + pub fn fields_pretty<'a>(&self) -> StrAssocArr<'a> { let last = ("aggs", self.agg_calls_pretty()); if self.group_key.count_ones(..) != 0 { let first = ("group_key", self.group_key_pretty()); @@ -636,6 +636,8 @@ impl Agg { } } +impl_distill_unit_from_fields!(Agg, stream::StreamPlanRef); + /// Rewritten version of [`AggCall`] which uses `InputRef` instead of `ExprImpl`. /// Refer to [`LogicalAggBuilder::try_rewrite_agg_call`] for more details. #[derive(Clone, PartialEq, Eq, Hash)] diff --git a/src/frontend/src/optimizer/plan_node/generic/hop_window.rs b/src/frontend/src/optimizer/plan_node/generic/hop_window.rs index a1c4c0799f18..2180d17469fc 100644 --- a/src/frontend/src/optimizer/plan_node/generic/hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/generic/hop_window.rs @@ -16,6 +16,7 @@ use std::fmt; use std::num::NonZeroUsize; use itertools::Itertools; +use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::{Field, Schema}; use risingwave_common::error::Result; use risingwave_common::types::{DataType, Interval, IntervalDisplay}; @@ -25,6 +26,7 @@ use risingwave_expr::ExprError; use super::super::utils::IndicesDisplay; use super::{GenericPlanNode, GenericPlanRef}; use crate::expr::{ExprImpl, ExprType, FunctionCall, InputRef, InputRefDisplay, Literal}; +use crate::impl_distill_unit_from_fields; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::plan_node::batch::BatchPlanRef; use crate::optimizer::property::{FunctionalDependencySet, Order}; @@ -338,9 +340,54 @@ impl HopWindow { } } + pub fn fields_pretty<'a>(&self) -> StrAssocArr<'a> { + let mut out = Vec::with_capacity(5); + let output_type = DataType::window_of(&self.time_col.data_type).unwrap(); + out.push(( + "time_col", + Pretty::display(&InputRefDisplay { + input_ref: &self.time_col, + input_schema: self.input.schema(), + }), + )); + out.push(("slide", Pretty::debug(&self.window_slide))); + out.push(("size", Pretty::debug(&self.window_size))); + if self + .output_indices + .iter() + .copied() + // Behavior is the same as `LogicalHopWindow::internal_column_num` + .eq(0..(self.input.schema().len() + 2)) + { + out.push(("output", Pretty::from("all"))); + } else { + let original_schema: Schema = self + .input + .schema() + .clone() + .into_fields() + .into_iter() + .chain([ + Field::with_name(output_type.clone(), "window_start"), + Field::with_name(output_type, "window_end"), + ]) + .collect(); + out.push(( + "output", + Pretty::display(&IndicesDisplay { + indices: &self.output_indices, + input_schema: &original_schema, + }), + )); + } + out + } + pub fn fmt_with_name(&self, f: &mut fmt::Formatter<'_>, name: &str) -> fmt::Result { let mut builder = f.debug_struct(name); self.fmt_fields_with_builder(&mut builder); builder.finish() } } + +impl_distill_unit_from_fields!(HopWindow, GenericPlanRef); diff --git a/src/frontend/src/optimizer/plan_node/generic/intersect.rs b/src/frontend/src/optimizer/plan_node/generic/intersect.rs index 8dfa870a5e54..995c02592df7 100644 --- a/src/frontend/src/optimizer/plan_node/generic/intersect.rs +++ b/src/frontend/src/optimizer/plan_node/generic/intersect.rs @@ -14,9 +14,11 @@ use std::fmt; +use pretty_xmlish::{StrAssocArr, Pretty}; use risingwave_common::catalog::Schema; use super::{GenericPlanNode, GenericPlanRef}; +use crate::impl_distill_unit_from_fields; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; @@ -56,4 +58,9 @@ impl Intersect { pub fn fmt_fields_with_builder(&self, builder: &mut fmt::DebugStruct<'_, '_>) { builder.field("all", &self.all); } + + pub fn fields_pretty<'a>(&self) -> StrAssocArr<'a> { + vec![("all", Pretty::debug(&self.all))] + } } +impl_distill_unit_from_fields!(Intersect, GenericPlanRef); diff --git a/src/frontend/src/optimizer/plan_node/generic/mod.rs b/src/frontend/src/optimizer/plan_node/generic/mod.rs index 481bb118ce91..861e565f6601 100644 --- a/src/frontend/src/optimizer/plan_node/generic/mod.rs +++ b/src/frontend/src/optimizer/plan_node/generic/mod.rs @@ -62,6 +62,18 @@ pub trait DistillUnit { fn distill_with_name<'a>(&self, name: &'a str) -> Pretty<'a>; } +#[macro_export] +macro_rules! impl_distill_unit_from_fields { + ($name:ident, $bound:path) => { + use crate::optimizer::plan_node::generic::DistillUnit; + impl DistillUnit for $name { + fn distill_with_name<'a>(&self, name: &'a str) -> Pretty<'a> { + Pretty::childless_record(name, self.fields_pretty()) + } + } + }; +} + pub trait GenericPlanRef { fn schema(&self) -> &Schema; fn logical_pk(&self) -> &[usize]; diff --git a/src/frontend/src/optimizer/plan_node/generic/project.rs b/src/frontend/src/optimizer/plan_node/generic/project.rs index 9c8818fc0b82..2b098966d918 100644 --- a/src/frontend/src/optimizer/plan_node/generic/project.rs +++ b/src/frontend/src/optimizer/plan_node/generic/project.rs @@ -17,7 +17,7 @@ use std::fmt; use std::fmt::Formatter; use fixedbitset::FixedBitSet; -use itertools::Itertools; +use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::{Field, Schema}; use risingwave_common::util::iter_util::ZipEqFast; @@ -178,26 +178,32 @@ impl Project { } pub fn fmt_fields_with_builder(&self, builder: &mut fmt::DebugStruct<'_, '_>, schema: &Schema) { - builder.field( - "exprs", - &self - .exprs - .iter() - .zip_eq_fast(schema.fields().iter()) - .map(|(expr, field)| AliasedExpr { - expr: ExprDisplay { - expr, - input_schema: self.input.schema(), - }, - alias: { - match expr { - ExprImpl::InputRef(_) | ExprImpl::Literal(_) => None, - _ => Some(field.name.clone()), - } - }, - }) - .collect_vec(), - ); + builder.field("exprs", &self.exprs_for_display(schema)); + } + + pub fn fields_pretty<'a>(&self, schema: &Schema) -> StrAssocArr<'a> { + let f = |t| Pretty::debug(&t); + let e = Pretty::Array(self.exprs_for_display(schema).iter().map(f).collect()); + vec![("exprs", e)] + } + + fn exprs_for_display<'a>(&'a self, schema: &Schema) -> Vec> { + self.exprs + .iter() + .zip_eq_fast(schema.fields().iter()) + .map(|(expr, field)| AliasedExpr { + expr: ExprDisplay { + expr, + input_schema: self.input.schema(), + }, + alias: { + match expr { + ExprImpl::InputRef(_) | ExprImpl::Literal(_) => None, + _ => Some(field.name.clone()), + } + }, + }) + .collect() } pub fn fmt_with_name( diff --git a/src/frontend/src/optimizer/plan_node/generic/union.rs b/src/frontend/src/optimizer/plan_node/generic/union.rs index 56dbce2141f3..f6a12289090a 100644 --- a/src/frontend/src/optimizer/plan_node/generic/union.rs +++ b/src/frontend/src/optimizer/plan_node/generic/union.rs @@ -14,6 +14,7 @@ use std::fmt; +use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::Schema; use super::{GenericPlanNode, GenericPlanRef}; @@ -72,4 +73,9 @@ impl Union { pub fn fmt_fields_with_builder(&self, builder: &mut fmt::DebugStruct<'_, '_>) { builder.field("all", &self.all); } + + pub fn fields_pretty<'a>(&self) -> StrAssocArr<'a> { + vec![("all", Pretty::debug(&self.all))] + } } +impl_distill_unit_from_fields!(Intersect, GenericPlanRef); From e2fa3c1784d0a2955b40d248b0249dec1b45c0b6 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 03:51:11 +0000 Subject: [PATCH 04/22] Fix --- src/frontend/src/optimizer/plan_node/generic/union.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/union.rs b/src/frontend/src/optimizer/plan_node/generic/union.rs index f6a12289090a..cdbe26bc3f2e 100644 --- a/src/frontend/src/optimizer/plan_node/generic/union.rs +++ b/src/frontend/src/optimizer/plan_node/generic/union.rs @@ -18,6 +18,7 @@ use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::Schema; use super::{GenericPlanNode, GenericPlanRef}; +use crate::impl_distill_unit_from_fields; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; @@ -78,4 +79,4 @@ impl Union { vec![("all", Pretty::debug(&self.all))] } } -impl_distill_unit_from_fields!(Intersect, GenericPlanRef); +impl_distill_unit_from_fields!(Union, GenericPlanRef); From 31553edd698fd52a64687935159a337bbb90eda9 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 03:51:30 +0000 Subject: [PATCH 05/22] Angry --- src/frontend/src/optimizer/plan_node/generic/agg.rs | 2 +- src/frontend/src/optimizer/plan_node/generic/intersect.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/agg.rs b/src/frontend/src/optimizer/plan_node/generic/agg.rs index be800a708995..9909cf9d058b 100644 --- a/src/frontend/src/optimizer/plan_node/generic/agg.rs +++ b/src/frontend/src/optimizer/plan_node/generic/agg.rs @@ -35,7 +35,7 @@ use crate::stream_fragmenter::BuildFragmentGraphState; use crate::utils::{ ColIndexMapping, ColIndexMappingRewriteExt, Condition, ConditionDisplay, IndexRewriter, }; -use crate::{TableCatalog, impl_distill_unit_from_fields}; +use crate::{impl_distill_unit_from_fields, TableCatalog}; /// [`Agg`] groups input data by their group key and computes aggregation functions. /// diff --git a/src/frontend/src/optimizer/plan_node/generic/intersect.rs b/src/frontend/src/optimizer/plan_node/generic/intersect.rs index 995c02592df7..2390253f0d97 100644 --- a/src/frontend/src/optimizer/plan_node/generic/intersect.rs +++ b/src/frontend/src/optimizer/plan_node/generic/intersect.rs @@ -14,7 +14,7 @@ use std::fmt; -use pretty_xmlish::{StrAssocArr, Pretty}; +use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::Schema; use super::{GenericPlanNode, GenericPlanRef}; From 39aabf39eac8b63cceed4da8dd607f2c69f1d3d8 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 03:52:08 +0000 Subject: [PATCH 06/22] More --- src/frontend/src/optimizer/plan_node/generic/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/mod.rs b/src/frontend/src/optimizer/plan_node/generic/mod.rs index 861e565f6601..5acc0b308ecc 100644 --- a/src/frontend/src/optimizer/plan_node/generic/mod.rs +++ b/src/frontend/src/optimizer/plan_node/generic/mod.rs @@ -65,7 +65,7 @@ pub trait DistillUnit { #[macro_export] macro_rules! impl_distill_unit_from_fields { ($name:ident, $bound:path) => { - use crate::optimizer::plan_node::generic::DistillUnit; + use $crate::optimizer::plan_node::generic::DistillUnit; impl DistillUnit for $name { fn distill_with_name<'a>(&self, name: &'a str) -> Pretty<'a> { Pretty::childless_record(name, self.fields_pretty()) From d076c87c8e607e7aa4b125b7df6139a30a78d714 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 03:57:29 +0000 Subject: [PATCH 07/22] Letting me down --- src/frontend/src/optimizer/plan_node/generic/agg.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/agg.rs b/src/frontend/src/optimizer/plan_node/generic/agg.rs index 9909cf9d058b..5e0de308b709 100644 --- a/src/frontend/src/optimizer/plan_node/generic/agg.rs +++ b/src/frontend/src/optimizer/plan_node/generic/agg.rs @@ -175,7 +175,7 @@ impl GenericPlanNode for Agg { } fn logical_pk(&self) -> Option> { - Some((0..self.group_key.count_ones(..)).collect_vec()) + Some((0..self.group_key.count_ones(..)).collect()) } fn ctx(&self) -> OptimizerContextRef { @@ -627,7 +627,7 @@ impl Agg { fn group_key_display(&self) -> Vec> { let f = |i| FieldDisplay(self.input.schema().fields.get(i).unwrap()); - self.group_key.ones().map(f).collect_vec() + self.group_key.ones().map(f).collect() } fn group_key_pretty<'a>(&self) -> Pretty<'a> { From f6c27f385c920b6c2448f4ec75412ab3e4324282 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:12:56 +0000 Subject: [PATCH 08/22] Return to nothing --- src/frontend/src/optimizer/plan_node/generic/agg.rs | 4 ++-- src/frontend/src/optimizer/plan_node/generic/hop_window.rs | 3 +-- src/frontend/src/optimizer/plan_node/generic/intersect.rs | 3 +-- src/frontend/src/optimizer/plan_node/generic/mod.rs | 2 +- src/frontend/src/optimizer/plan_node/generic/union.rs | 3 +-- 5 files changed, 6 insertions(+), 9 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/agg.rs b/src/frontend/src/optimizer/plan_node/generic/agg.rs index 5e0de308b709..03fd443a9814 100644 --- a/src/frontend/src/optimizer/plan_node/generic/agg.rs +++ b/src/frontend/src/optimizer/plan_node/generic/agg.rs @@ -26,7 +26,7 @@ use risingwave_pb::expr::PbAggCall; use risingwave_pb::stream_plan::{agg_call_state, AggCallState as AggCallStatePb}; use super::super::utils::TableCatalogBuilder; -use super::{stream, GenericPlanNode, GenericPlanRef}; +use super::{impl_distill_unit_from_fields, stream, GenericPlanNode, GenericPlanRef}; use crate::expr::{Expr, ExprRewriter, InputRef, InputRefDisplay}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::plan_node::batch::BatchPlanRef; @@ -35,7 +35,7 @@ use crate::stream_fragmenter::BuildFragmentGraphState; use crate::utils::{ ColIndexMapping, ColIndexMappingRewriteExt, Condition, ConditionDisplay, IndexRewriter, }; -use crate::{impl_distill_unit_from_fields, TableCatalog}; +use crate::TableCatalog; /// [`Agg`] groups input data by their group key and computes aggregation functions. /// diff --git a/src/frontend/src/optimizer/plan_node/generic/hop_window.rs b/src/frontend/src/optimizer/plan_node/generic/hop_window.rs index 2180d17469fc..b66320391222 100644 --- a/src/frontend/src/optimizer/plan_node/generic/hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/generic/hop_window.rs @@ -24,9 +24,8 @@ use risingwave_common::util::column_index_mapping::ColIndexMapping; use risingwave_expr::ExprError; use super::super::utils::IndicesDisplay; -use super::{GenericPlanNode, GenericPlanRef}; +use super::{GenericPlanNode, GenericPlanRef, impl_distill_unit_from_fields}; use crate::expr::{ExprImpl, ExprType, FunctionCall, InputRef, InputRefDisplay, Literal}; -use crate::impl_distill_unit_from_fields; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::plan_node::batch::BatchPlanRef; use crate::optimizer::property::{FunctionalDependencySet, Order}; diff --git a/src/frontend/src/optimizer/plan_node/generic/intersect.rs b/src/frontend/src/optimizer/plan_node/generic/intersect.rs index 2390253f0d97..dbc09a4dbaf9 100644 --- a/src/frontend/src/optimizer/plan_node/generic/intersect.rs +++ b/src/frontend/src/optimizer/plan_node/generic/intersect.rs @@ -17,8 +17,7 @@ use std::fmt; use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::Schema; -use super::{GenericPlanNode, GenericPlanRef}; -use crate::impl_distill_unit_from_fields; +use super::{impl_distill_unit_from_fields, GenericPlanNode, GenericPlanRef}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; diff --git a/src/frontend/src/optimizer/plan_node/generic/mod.rs b/src/frontend/src/optimizer/plan_node/generic/mod.rs index 5acc0b308ecc..ad2ac6069f37 100644 --- a/src/frontend/src/optimizer/plan_node/generic/mod.rs +++ b/src/frontend/src/optimizer/plan_node/generic/mod.rs @@ -62,7 +62,6 @@ pub trait DistillUnit { fn distill_with_name<'a>(&self, name: &'a str) -> Pretty<'a>; } -#[macro_export] macro_rules! impl_distill_unit_from_fields { ($name:ident, $bound:path) => { use $crate::optimizer::plan_node::generic::DistillUnit; @@ -73,6 +72,7 @@ macro_rules! impl_distill_unit_from_fields { } }; } +pub(super) use impl_distill_unit_from_fields; pub trait GenericPlanRef { fn schema(&self) -> &Schema; diff --git a/src/frontend/src/optimizer/plan_node/generic/union.rs b/src/frontend/src/optimizer/plan_node/generic/union.rs index cdbe26bc3f2e..f0a965d777b5 100644 --- a/src/frontend/src/optimizer/plan_node/generic/union.rs +++ b/src/frontend/src/optimizer/plan_node/generic/union.rs @@ -17,8 +17,7 @@ use std::fmt; use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::Schema; -use super::{GenericPlanNode, GenericPlanRef}; -use crate::impl_distill_unit_from_fields; +use super::{impl_distill_unit_from_fields, GenericPlanNode, GenericPlanRef}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; From bef686cbc14557fc4af30b9cdb9bf827808e7b67 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:30:54 +0000 Subject: [PATCH 09/22] Filter --- .../src/optimizer/plan_node/generic/filter.rs | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/filter.rs b/src/frontend/src/optimizer/plan_node/generic/filter.rs index 7d24a4939eca..35bfdd53d0ab 100644 --- a/src/frontend/src/optimizer/plan_node/generic/filter.rs +++ b/src/frontend/src/optimizer/plan_node/generic/filter.rs @@ -14,9 +14,10 @@ use std::fmt; +use pretty_xmlish::Pretty; use risingwave_common::catalog::Schema; -use super::{GenericPlanNode, GenericPlanRef}; +use super::{DistillUnit, GenericPlanNode, GenericPlanRef}; use crate::expr::ExprRewriter; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; @@ -32,6 +33,17 @@ pub struct Filter { pub input: PlanRef, } +impl DistillUnit for Filter { + fn distill_with_name<'a>(&self, name: &'a str) -> Pretty<'a> { + let input_schema = self.input.schema(); + let predicate = ConditionDisplay { + condition: &self.predicate, + input_schema, + }; + Pretty::childless_record(name, vec![("predicate", Pretty::display(&predicate))]) + } +} + impl Filter { pub(crate) fn fmt_with_name(&self, f: &mut fmt::Formatter<'_>, name: &str) -> fmt::Result { let input_schema = self.input.schema(); From d52614c43234a41a6dd121d6e6e1d1299466e2ac Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:31:23 +0000 Subject: [PATCH 10/22] Fmt --- src/frontend/src/optimizer/plan_node/generic/hop_window.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/hop_window.rs b/src/frontend/src/optimizer/plan_node/generic/hop_window.rs index b66320391222..7f96defa5d22 100644 --- a/src/frontend/src/optimizer/plan_node/generic/hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/generic/hop_window.rs @@ -24,7 +24,7 @@ use risingwave_common::util::column_index_mapping::ColIndexMapping; use risingwave_expr::ExprError; use super::super::utils::IndicesDisplay; -use super::{GenericPlanNode, GenericPlanRef, impl_distill_unit_from_fields}; +use super::{impl_distill_unit_from_fields, GenericPlanNode, GenericPlanRef}; use crate::expr::{ExprImpl, ExprType, FunctionCall, InputRef, InputRefDisplay, Literal}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::plan_node::batch::BatchPlanRef; From 320939b0a064885b87a080a1ecd5f947222f831b Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:44:14 +0000 Subject: [PATCH 11/22] Progress --- .../src/optimizer/plan_node/batch_delete.rs | 9 ++---- .../src/optimizer/plan_node/batch_exchange.rs | 31 +++++++++++++++++-- .../src/optimizer/plan_node/batch_expand.rs | 7 ++--- .../src/optimizer/plan_node/batch_filter.rs | 8 ++--- .../optimizer/plan_node/batch_group_topn.rs | 7 ++--- .../src/optimizer/plan_node/batch_hash_agg.rs | 7 ++--- .../src/optimizer/plan_node/logical_agg.rs | 7 ++--- .../src/optimizer/plan_node/logical_dedup.rs | 7 ++--- .../src/optimizer/plan_node/logical_delete.rs | 8 ++--- .../src/optimizer/plan_node/logical_except.rs | 7 ++--- .../src/optimizer/plan_node/logical_expand.rs | 8 ++--- .../src/optimizer/plan_node/logical_filter.rs | 8 ++--- .../optimizer/plan_node/logical_hop_window.rs | 8 ++--- .../optimizer/plan_node/logical_intersect.rs | 7 ++--- .../plan_node/logical_over_window.rs | 8 ++--- .../optimizer/plan_node/logical_project.rs | 9 +++++- .../src/optimizer/plan_node/logical_topn.rs | 7 ++--- src/frontend/src/optimizer/plan_node/utils.rs | 26 ++++++++++++++++ 18 files changed, 93 insertions(+), 86 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/batch_delete.rs b/src/frontend/src/optimizer/plan_node/batch_delete.rs index f9c233eb61cd..a7344e98e9d2 100644 --- a/src/frontend/src/optimizer/plan_node/batch_delete.rs +++ b/src/frontend/src/optimizer/plan_node/batch_delete.rs @@ -12,12 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; - use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::DeleteNode; +use super::utils::impl_distill_by_unit; use super::{ generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -43,11 +42,7 @@ impl BatchDelete { } } -impl fmt::Display for BatchDelete { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchDelete") - } -} +impl_distill_by_unit!(BatchDelete, logical, "BatchDelete"); impl PlanTreeNodeUnary for BatchDelete { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_exchange.rs b/src/frontend/src/optimizer/plan_node/batch_exchange.rs index d69ae041f65b..34a27d843bdc 100644 --- a/src/frontend/src/optimizer/plan_node/batch_exchange.rs +++ b/src/frontend/src/optimizer/plan_node/batch_exchange.rs @@ -14,10 +14,12 @@ use std::fmt; +use pretty_xmlish::Pretty; use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::{ExchangeNode, MergeSortExchangeNode}; +use super::utils::Distill; use super::{ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch}; use crate::optimizer::plan_node::ToLocalBatch; use crate::optimizer::property::{Distribution, DistributionDisplay, Order, OrderDisplay}; @@ -41,20 +43,45 @@ impl BatchExchange { impl fmt::Display for BatchExchange { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let input_schema = self.input.schema(); write!( f, "BatchExchange {{ order: {}, dist: {} }}", OrderDisplay { order: &self.base.order, - input_schema: self.input.schema() + input_schema }, DistributionDisplay { distribution: &self.base.dist, - input_schema: self.input.schema() + input_schema } ) } } +impl Distill for BatchExchange { + fn distill<'a>(&self) -> Pretty<'a> { + let input_schema = self.input.schema(); + Pretty::childless_record( + "BatchExchange", + vec![ + ( + "order", + Pretty::display(&OrderDisplay { + order: &self.base.order, + input_schema, + }), + ), + ( + "dist", + Pretty::display(&DistributionDisplay { + distribution: &self.base.dist, + input_schema, + }), + ), + ], + ) + } +} impl PlanTreeNodeUnary for BatchExchange { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_expand.rs b/src/frontend/src/optimizer/plan_node/batch_expand.rs index f81b1d7e8463..de213a30500f 100644 --- a/src/frontend/src/optimizer/plan_node/batch_expand.rs +++ b/src/frontend/src/optimizer/plan_node/batch_expand.rs @@ -20,6 +20,7 @@ use risingwave_pb::batch_plan::expand_node::Subset; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::ExpandNode; +use super::utils::impl_distill_by_unit; use super::{generic, ExprRewritable}; use crate::optimizer::plan_node::{ PlanBase, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, ToLocalBatch, @@ -51,11 +52,7 @@ impl BatchExpand { } } -impl fmt::Display for BatchExpand { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchExpand") - } -} +impl_distill_by_unit!(BatchExpand, logical, "BatchExpand"); impl PlanTreeNodeUnary for BatchExpand { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_filter.rs b/src/frontend/src/optimizer/plan_node/batch_filter.rs index 476bbe4f1fcc..052b6b5398ba 100644 --- a/src/frontend/src/optimizer/plan_node/batch_filter.rs +++ b/src/frontend/src/optimizer/plan_node/batch_filter.rs @@ -18,6 +18,7 @@ use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::FilterNode; +use super::utils::impl_distill_by_unit; use super::{generic, ExprRewritable, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch}; use crate::expr::{Expr, ExprImpl, ExprRewriter}; use crate::optimizer::plan_node::{PlanBase, ToLocalBatch}; @@ -45,12 +46,7 @@ impl BatchFilter { &self.logical.predicate } } - -impl fmt::Display for BatchFilter { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchFilter") - } -} +impl_distill_by_unit!(BatchFilter, logical, "BatchFilter"); impl PlanTreeNodeUnary for BatchFilter { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_group_topn.rs b/src/frontend/src/optimizer/plan_node/batch_group_topn.rs index 21e137c0da19..889d9385e162 100644 --- a/src/frontend/src/optimizer/plan_node/batch_group_topn.rs +++ b/src/frontend/src/optimizer/plan_node/batch_group_topn.rs @@ -18,6 +18,7 @@ use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::GroupTopNNode; +use super::utils::impl_distill_by_unit; use super::{ generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -47,11 +48,7 @@ impl BatchGroupTopN { } } -impl fmt::Display for BatchGroupTopN { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchGroupTopN") - } -} +impl_distill_by_unit!(BatchGroupTopN, logical, "BatchGroupTopN"); impl PlanTreeNodeUnary for BatchGroupTopN { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs b/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs index bc573d339616..7e0b21b8d7ba 100644 --- a/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs @@ -21,6 +21,7 @@ use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::HashAggNode; use super::generic::{self, GenericPlanRef, PlanAggCall}; +use super::utils::impl_distill_by_unit; use super::{ ExprRewritable, PlanBase, PlanNodeType, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, @@ -100,11 +101,7 @@ impl BatchHashAgg { } } -impl fmt::Display for BatchHashAgg { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchHashAgg") - } -} +impl_distill_by_unit!(BatchHashAgg, logical, "BatchHashAgg"); impl PlanTreeNodeUnary for BatchHashAgg { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_agg.rs b/src/frontend/src/optimizer/plan_node/logical_agg.rs index f1f33367efa9..7759029f8213 100644 --- a/src/frontend/src/optimizer/plan_node/logical_agg.rs +++ b/src/frontend/src/optimizer/plan_node/logical_agg.rs @@ -22,6 +22,7 @@ use risingwave_common::util::sort_util::ColumnOrder; use risingwave_expr::agg::AggKind; use super::generic::{self, Agg, GenericPlanRef, PlanAggCall, ProjectBuilder}; +use super::utils::impl_distill_by_unit; use super::{ BatchHashAgg, BatchSimpleAgg, ColPrunable, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, StreamHashAgg, StreamProject, StreamSimpleAgg, @@ -862,11 +863,7 @@ impl PlanTreeNodeUnary for LogicalAgg { impl_plan_tree_node_for_unary! {LogicalAgg} -impl fmt::Display for LogicalAgg { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalAgg") - } -} +impl_distill_by_unit!(LogicalAgg, core, "LogicalAgg"); impl ExprRewritable for LogicalAgg { fn has_rewritable_expr(&self) -> bool { diff --git a/src/frontend/src/optimizer/plan_node/logical_dedup.rs b/src/frontend/src/optimizer/plan_node/logical_dedup.rs index 28b20de15caa..daa26174231e 100644 --- a/src/frontend/src/optimizer/plan_node/logical_dedup.rs +++ b/src/frontend/src/optimizer/plan_node/logical_dedup.rs @@ -20,6 +20,7 @@ use risingwave_common::error::Result; use risingwave_common::util::column_index_mapping::ColIndexMapping; use super::generic::Limit; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, generic, BatchGroupTopN, ColPrunable, ColumnPruningContext, ExprRewritable, LogicalProject, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, @@ -183,8 +184,4 @@ impl ColPrunable for LogicalDedup { } } -impl fmt::Display for LogicalDedup { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalDedup") - } -} +impl_distill_by_unit!(LogicalDedup, core, "LogicalDedup"); diff --git a/src/frontend/src/optimizer/plan_node/logical_delete.rs b/src/frontend/src/optimizer/plan_node/logical_delete.rs index c1d4d6c7244a..2b16e71c793e 100644 --- a/src/frontend/src/optimizer/plan_node/logical_delete.rs +++ b/src/frontend/src/optimizer/plan_node/logical_delete.rs @@ -18,6 +18,7 @@ use risingwave_common::catalog::{Field, Schema, TableVersionId}; use risingwave_common::error::Result; use risingwave_common::types::DataType; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, generic, BatchDelete, ColPrunable, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, ToBatch, ToStream, @@ -79,12 +80,7 @@ impl PlanTreeNodeUnary for LogicalDelete { } impl_plan_tree_node_for_unary! { LogicalDelete } - -impl fmt::Display for LogicalDelete { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalDelete") - } -} +impl_distill_by_unit!(LogicalDelete, core, "LogicalDelete"); impl ColPrunable for LogicalDelete { fn prune_col(&self, _required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_except.rs b/src/frontend/src/optimizer/plan_node/logical_except.rs index 3b77908b4ea0..15bc823c3186 100644 --- a/src/frontend/src/optimizer/plan_node/logical_except.rs +++ b/src/frontend/src/optimizer/plan_node/logical_except.rs @@ -18,6 +18,7 @@ use itertools::Itertools; use risingwave_common::catalog::Schema; use risingwave_common::error::Result; +use super::utils::impl_distill_by_unit; use super::{ColPrunable, ExprRewritable, PlanBase, PlanRef, PredicatePushdown, ToBatch, ToStream}; use crate::optimizer::plan_node::generic::GenericPlanRef; use crate::optimizer::plan_node::{ @@ -65,11 +66,7 @@ impl PlanTreeNode for LogicalExcept { } } -impl fmt::Display for LogicalExcept { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.fmt_with_name(f, "LogicalExcept") - } -} +impl_distill_by_unit!(LogicalExcept, core, "LogicalExcept"); impl ColPrunable for LogicalExcept { fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_expand.rs b/src/frontend/src/optimizer/plan_node/logical_expand.rs index b984bcd941f8..acb2abf89181 100644 --- a/src/frontend/src/optimizer/plan_node/logical_expand.rs +++ b/src/frontend/src/optimizer/plan_node/logical_expand.rs @@ -17,6 +17,7 @@ use std::fmt; use itertools::Itertools; use risingwave_common::error::Result; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, generic, BatchExpand, ColPrunable, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, StreamExpand, ToBatch, ToStream, @@ -110,12 +111,7 @@ impl PlanTreeNodeUnary for LogicalExpand { } impl_plan_tree_node_for_unary! {LogicalExpand} - -impl fmt::Display for LogicalExpand { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.fmt_with_name(f, "LogicalExpand") - } -} +impl_distill_by_unit!(LogicalExpand, core, "LogicalExpand"); impl ColPrunable for LogicalExpand { fn prune_col(&self, _required_cols: &[usize], _ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_filter.rs b/src/frontend/src/optimizer/plan_node/logical_filter.rs index 843ece69388f..cb38fc146dd3 100644 --- a/src/frontend/src/optimizer/plan_node/logical_filter.rs +++ b/src/frontend/src/optimizer/plan_node/logical_filter.rs @@ -20,6 +20,7 @@ use risingwave_common::bail; use risingwave_common::error::Result; use risingwave_common::types::DataType; +use super::utils::impl_distill_by_unit; use super::{ generic, ColPrunable, ExprRewritable, LogicalProject, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, ToBatch, ToStream, @@ -117,12 +118,7 @@ impl PlanTreeNodeUnary for LogicalFilter { } impl_plan_tree_node_for_unary! {LogicalFilter} - -impl fmt::Display for LogicalFilter { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalFilter") - } -} +impl_distill_by_unit!(LogicalFilter, core, "LogicalFilter"); impl ColPrunable for LogicalFilter { fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_hop_window.rs b/src/frontend/src/optimizer/plan_node/logical_hop_window.rs index 0bd9d8d8c9f2..681ad8ab0522 100644 --- a/src/frontend/src/optimizer/plan_node/logical_hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/logical_hop_window.rs @@ -20,6 +20,7 @@ use risingwave_common::error::Result; use risingwave_common::types::Interval; use super::generic::GenericPlanNode; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, generic, BatchHopWindow, ColPrunable, ExprRewritable, LogicalFilter, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, StreamHopWindow, ToBatch, ToStream, @@ -217,12 +218,7 @@ impl PlanTreeNodeUnary for LogicalHopWindow { } impl_plan_tree_node_for_unary! {LogicalHopWindow} - -impl fmt::Display for LogicalHopWindow { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalHopWindow") - } -} +impl_distill_by_unit!(LogicalHopWindow, core, "LogicalHopWindow"); impl ColPrunable for LogicalHopWindow { fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_intersect.rs b/src/frontend/src/optimizer/plan_node/logical_intersect.rs index 909ea90dcffb..6c92edb0bbcf 100644 --- a/src/frontend/src/optimizer/plan_node/logical_intersect.rs +++ b/src/frontend/src/optimizer/plan_node/logical_intersect.rs @@ -18,6 +18,7 @@ use itertools::Itertools; use risingwave_common::catalog::Schema; use risingwave_common::error::Result; +use super::utils::impl_distill_by_unit; use super::{ColPrunable, ExprRewritable, PlanBase, PlanRef, PredicatePushdown, ToBatch, ToStream}; use crate::optimizer::plan_node::{ generic, ColumnPruningContext, PlanTreeNode, PredicatePushdownContext, RewriteStreamContext, @@ -68,11 +69,7 @@ impl PlanTreeNode for LogicalIntersect { } } -impl fmt::Display for LogicalIntersect { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.fmt_with_name(f, "LogicalIntersect") - } -} +impl_distill_by_unit!(LogicalIntersect, core, "LogicalIntersect"); impl ColPrunable for LogicalIntersect { fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_over_window.rs b/src/frontend/src/optimizer/plan_node/logical_over_window.rs index 9a6e5093ce72..34fb1ea34446 100644 --- a/src/frontend/src/optimizer/plan_node/logical_over_window.rs +++ b/src/frontend/src/optimizer/plan_node/logical_over_window.rs @@ -23,6 +23,7 @@ use risingwave_expr::agg::AggKind; use risingwave_expr::function::window::{Frame, FrameBound, WindowFuncKind}; use super::generic::{GenericPlanRef, OverWindow, PlanWindowFunction, ProjectBuilder}; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, ColPrunable, ExprRewritable, LogicalProject, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, StreamEowcOverWindow, StreamSort, ToBatch, ToStream, @@ -558,12 +559,7 @@ impl PlanTreeNodeUnary for LogicalOverWindow { } impl_plan_tree_node_for_unary! { LogicalOverWindow } - -impl fmt::Display for LogicalOverWindow { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalOverWindow") - } -} +impl_distill_by_unit!(LogicalOverWindow, core, "LogicalOverWindow"); impl ColPrunable for LogicalOverWindow { fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_project.rs b/src/frontend/src/optimizer/plan_node/logical_project.rs index 51147dd0bb6b..7771897ff325 100644 --- a/src/frontend/src/optimizer/plan_node/logical_project.rs +++ b/src/frontend/src/optimizer/plan_node/logical_project.rs @@ -16,8 +16,10 @@ use std::fmt; use fixedbitset::FixedBitSet; use itertools::Itertools; +use pretty_xmlish::Pretty; use risingwave_common::error::Result; +use super::utils::Distill; use super::{ gen_filter_and_pushdown, generic, BatchProject, ColPrunable, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, StreamProject, ToBatch, ToStream, @@ -136,7 +138,12 @@ impl_plan_tree_node_for_unary! {LogicalProject} impl fmt::Display for LogicalProject { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.fmt_with_name(f, "LogicalProject") + self.core.fmt_with_name(f, "LogicalProject", self.base.schema()) + } +} +impl Distill for LogicalProject { + fn distill<'a>(&self) -> Pretty<'a> { + self.core.distill_with_name("LogicalProject", self.base.schema()) } } diff --git a/src/frontend/src/optimizer/plan_node/logical_topn.rs b/src/frontend/src/optimizer/plan_node/logical_topn.rs index e4b27fd397bd..03707622811e 100644 --- a/src/frontend/src/optimizer/plan_node/logical_topn.rs +++ b/src/frontend/src/optimizer/plan_node/logical_topn.rs @@ -20,6 +20,7 @@ use risingwave_common::error::{ErrorCode, Result, RwError}; use risingwave_common::util::sort_util::ColumnOrder; use super::generic::Limit; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, generic, BatchGroupTopN, ColPrunable, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, StreamGroupTopN, StreamProject, ToBatch, @@ -209,11 +210,7 @@ impl PlanTreeNodeUnary for LogicalTopN { } } impl_plan_tree_node_for_unary! {LogicalTopN} -impl fmt::Display for LogicalTopN { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalTopN") - } -} +impl_distill_by_unit!(LogicalTopN, core, "LogicalTopN"); impl ColPrunable for LogicalTopN { fn prune_col(&self, required_cols: &[usize], ctx: &mut ColumnPruningContext) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/utils.rs b/src/frontend/src/optimizer/plan_node/utils.rs index fd827a86862e..04e869dc8b83 100644 --- a/src/frontend/src/optimizer/plan_node/utils.rs +++ b/src/frontend/src/optimizer/plan_node/utils.rs @@ -17,6 +17,7 @@ use std::{fmt, vec}; use fixedbitset::FixedBitSet; use itertools::Itertools; +use pretty_xmlish::Pretty; use risingwave_common::catalog::{ColumnCatalog, ColumnDesc, ConflictBehavior, Field, Schema}; use risingwave_common::util::sort_util::{ColumnOrder, OrderType}; @@ -159,6 +160,31 @@ impl TableCatalogBuilder { } } +/// See also [`super::generic::DistillUnit`]. +pub trait Distill { + fn distill<'a>(&self) -> Pretty<'a>; +} + +macro_rules! impl_distill_by_unit { + ($ty:ty, $core:ident, $name:expr) => { + use pretty_xmlish::Pretty; + use $crate::optimizer::plan_node::generic::DistillUnit; + use $crate::optimizer::plan_node::utils::Distill; + impl Distill for $ty { + fn distill<'a>(&self) -> Pretty<'a> { + self.$core.distill_with_name($name) + } + } + + impl std::fmt::Display for $ty { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.$core.fmt_with_name(f, $name) + } + } + }; +} +pub(crate) use impl_distill_by_unit; + #[derive(Clone, Copy)] pub struct IndicesDisplay<'a> { pub indices: &'a [usize], From 9885e9465fedef3beb393658cb8ddb08442523e0 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:45:09 +0000 Subject: [PATCH 12/22] Ute! --- src/frontend/src/optimizer/plan_node/logical_project.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/logical_project.rs b/src/frontend/src/optimizer/plan_node/logical_project.rs index 7771897ff325..508beb6ff38b 100644 --- a/src/frontend/src/optimizer/plan_node/logical_project.rs +++ b/src/frontend/src/optimizer/plan_node/logical_project.rs @@ -25,7 +25,7 @@ use super::{ PlanTreeNodeUnary, PredicatePushdown, StreamProject, ToBatch, ToStream, }; use crate::expr::{collect_input_refs, ExprImpl, ExprRewriter, InputRef}; -use crate::optimizer::plan_node::generic::GenericPlanRef; +use crate::optimizer::plan_node::generic::{DistillUnit, GenericPlanRef}; use crate::optimizer::plan_node::{ ColumnPruningContext, PredicatePushdownContext, RewriteStreamContext, ToStreamContext, }; @@ -138,12 +138,14 @@ impl_plan_tree_node_for_unary! {LogicalProject} impl fmt::Display for LogicalProject { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalProject", self.base.schema()) + self.core + .fmt_with_name(f, "LogicalProject", self.base.schema()) } } impl Distill for LogicalProject { fn distill<'a>(&self) -> Pretty<'a> { - self.core.distill_with_name("LogicalProject", self.base.schema()) + self.core + .distill_with_name("LogicalProject", self.base.schema()) } } From 641986f8b4ac435f597c92a4dd6a99de4bb724ec Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:48:55 +0000 Subject: [PATCH 13/22] So? --- src/frontend/src/optimizer/plan_node/generic/project.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_project.rs | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/project.rs b/src/frontend/src/optimizer/plan_node/generic/project.rs index 2b098966d918..521771f1857e 100644 --- a/src/frontend/src/optimizer/plan_node/generic/project.rs +++ b/src/frontend/src/optimizer/plan_node/generic/project.rs @@ -21,7 +21,7 @@ use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::{Field, Schema}; use risingwave_common::util::iter_util::ZipEqFast; -use super::{GenericPlanNode, GenericPlanRef}; +use super::{GenericPlanNode, GenericPlanRef, impl_distill_unit_from_fields}; use crate::expr::{assert_input_ref, Expr, ExprDisplay, ExprImpl, ExprRewriter, InputRef}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; diff --git a/src/frontend/src/optimizer/plan_node/logical_project.rs b/src/frontend/src/optimizer/plan_node/logical_project.rs index 508beb6ff38b..6fe21375e251 100644 --- a/src/frontend/src/optimizer/plan_node/logical_project.rs +++ b/src/frontend/src/optimizer/plan_node/logical_project.rs @@ -144,8 +144,10 @@ impl fmt::Display for LogicalProject { } impl Distill for LogicalProject { fn distill<'a>(&self) -> Pretty<'a> { - self.core - .distill_with_name("LogicalProject", self.base.schema()) + Pretty::childless_record( + "LogicalProject", + self.core.fields_pretty(self.base.schema()), + ) } } From d35aba6d7b01df4cceef71bec5cbbf7301c91e8c Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:49:07 +0000 Subject: [PATCH 14/22] So --- src/frontend/src/optimizer/plan_node/generic/project.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/frontend/src/optimizer/plan_node/generic/project.rs b/src/frontend/src/optimizer/plan_node/generic/project.rs index 521771f1857e..10024aeb4d0c 100644 --- a/src/frontend/src/optimizer/plan_node/generic/project.rs +++ b/src/frontend/src/optimizer/plan_node/generic/project.rs @@ -21,7 +21,7 @@ use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::{Field, Schema}; use risingwave_common::util::iter_util::ZipEqFast; -use super::{GenericPlanNode, GenericPlanRef, impl_distill_unit_from_fields}; +use super::{impl_distill_unit_from_fields, GenericPlanNode, GenericPlanRef}; use crate::expr::{assert_input_ref, Expr, ExprDisplay, ExprImpl, ExprRewriter, InputRef}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; From 788f9b9d3bb17c618519c7d74fb9910cdbe99c27 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:54:39 +0000 Subject: [PATCH 15/22] More --- src/frontend/src/optimizer/plan_node/logical_update.rs | 8 ++------ src/frontend/src/optimizer/plan_node/stream_dedup.rs | 10 +++------- .../src/optimizer/plan_node/stream_eowc_over_window.rs | 8 ++------ src/frontend/src/optimizer/plan_node/stream_expand.rs | 7 ++----- src/frontend/src/optimizer/plan_node/stream_filter.rs | 7 ++----- .../src/optimizer/plan_node/stream_project_set.rs | 8 ++------ 6 files changed, 13 insertions(+), 35 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/logical_update.rs b/src/frontend/src/optimizer/plan_node/logical_update.rs index 556874f2d962..7a10ff500ea8 100644 --- a/src/frontend/src/optimizer/plan_node/logical_update.rs +++ b/src/frontend/src/optimizer/plan_node/logical_update.rs @@ -18,6 +18,7 @@ use risingwave_common::catalog::{Field, Schema, TableVersionId}; use risingwave_common::error::Result; use risingwave_common::types::DataType; +use super::utils::impl_distill_by_unit; use super::{ gen_filter_and_pushdown, generic, BatchUpdate, ColPrunable, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, PredicatePushdown, ToBatch, ToStream, @@ -86,12 +87,7 @@ impl PlanTreeNodeUnary for LogicalUpdate { } impl_plan_tree_node_for_unary! { LogicalUpdate } - -impl fmt::Display for LogicalUpdate { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.core.fmt_with_name(f, "LogicalUpdate") - } -} +impl_distill_by_unit!(LogicalUpdate, core, "LogicalUpdate"); impl ExprRewritable for LogicalUpdate { fn has_rewritable_expr(&self) -> bool { diff --git a/src/frontend/src/optimizer/plan_node/stream_dedup.rs b/src/frontend/src/optimizer/plan_node/stream_dedup.rs index d9abd7b9af87..e2945b6a1500 100644 --- a/src/frontend/src/optimizer/plan_node/stream_dedup.rs +++ b/src/frontend/src/optimizer/plan_node/stream_dedup.rs @@ -20,7 +20,7 @@ use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::DedupNode; use super::generic::{self, GenericPlanNode, GenericPlanRef}; -use super::utils::TableCatalogBuilder; +use super::utils::{TableCatalogBuilder, impl_distill_by_unit}; use super::{ExprRewritable, PlanBase, PlanTreeNodeUnary, StreamNode}; use crate::optimizer::plan_node::stream::StreamPlanRef; use crate::optimizer::plan_node::PlanRef; @@ -71,12 +71,8 @@ impl StreamDedup { } } -impl fmt::Display for StreamDedup { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - assert!(self.base.append_only()); - self.logical.fmt_with_name(f, "StreamAppendOnlyDedup") - } -} +// assert!(self.base.append_only()); +impl_distill_by_unit!(StreamDedup, logical, "StreamAppendOnlyDedup"); impl PlanTreeNodeUnary for StreamDedup { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs b/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs index 13f59b02504a..30816f7dfbfb 100644 --- a/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs +++ b/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs @@ -20,7 +20,7 @@ use risingwave_common::util::sort_util::OrderType; use risingwave_pb::stream_plan::stream_node::PbNodeBody; use super::generic::{self, PlanWindowFunction}; -use super::utils::TableCatalogBuilder; +use super::utils::{TableCatalogBuilder, impl_distill_by_unit}; use super::{ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, StreamNode}; use crate::optimizer::plan_node::stream::StreamPlanRef; use crate::stream_fragmenter::BuildFragmentGraphState; @@ -114,11 +114,7 @@ impl StreamEowcOverWindow { } } -impl fmt::Display for StreamEowcOverWindow { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "StreamEowcOverWindow") - } -} +impl_distill_by_unit!(StreamEowcOverWindow, logical, "StreamEowcOverWindow"); impl PlanTreeNodeUnary for StreamEowcOverWindow { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/stream_expand.rs b/src/frontend/src/optimizer/plan_node/stream_expand.rs index dbcec9fa9349..bca10ed77834 100644 --- a/src/frontend/src/optimizer/plan_node/stream_expand.rs +++ b/src/frontend/src/optimizer/plan_node/stream_expand.rs @@ -20,6 +20,7 @@ use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::ExpandNode; use super::stream::StreamPlanRef; +use super::utils::impl_distill_by_unit; use super::{generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, StreamNode}; use crate::optimizer::property::Distribution; use crate::stream_fragmenter::BuildFragmentGraphState; @@ -66,11 +67,7 @@ impl StreamExpand { } } -impl fmt::Display for StreamExpand { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "StreamExpand") - } -} +impl_distill_by_unit!(StreamExpand, logical, "StreamExpand"); impl PlanTreeNodeUnary for StreamExpand { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/stream_filter.rs b/src/frontend/src/optimizer/plan_node/stream_filter.rs index 5581ebfe8d27..93b4b1787795 100644 --- a/src/frontend/src/optimizer/plan_node/stream_filter.rs +++ b/src/frontend/src/optimizer/plan_node/stream_filter.rs @@ -18,6 +18,7 @@ use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::FilterNode; use super::stream::StreamPlanRef; +use super::utils::impl_distill_by_unit; use super::{generic, ExprRewritable, PlanRef, PlanTreeNodeUnary, StreamNode}; use crate::expr::{Expr, ExprImpl, ExprRewriter}; use crate::optimizer::plan_node::PlanBase; @@ -51,11 +52,7 @@ impl StreamFilter { } } -impl fmt::Display for StreamFilter { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "StreamFilter") - } -} +impl_distill_by_unit!(StreamFilter, logical, "StreamFilter"); impl PlanTreeNodeUnary for StreamFilter { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/stream_project_set.rs b/src/frontend/src/optimizer/plan_node/stream_project_set.rs index 0e5c19937e08..a673bb830b36 100644 --- a/src/frontend/src/optimizer/plan_node/stream_project_set.rs +++ b/src/frontend/src/optimizer/plan_node/stream_project_set.rs @@ -20,6 +20,7 @@ use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::ProjectSetNode; use super::stream::StreamPlanRef; +use super::utils::impl_distill_by_unit; use super::{generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, StreamNode}; use crate::expr::{try_derive_watermark, ExprRewriter}; use crate::stream_fragmenter::BuildFragmentGraphState; @@ -60,12 +61,7 @@ impl StreamProjectSet { StreamProjectSet { base, logical } } } - -impl fmt::Display for StreamProjectSet { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "StreamProjectSet") - } -} +impl_distill_by_unit!(StreamProjectSet, logical, "StreamProjectSet"); impl PlanTreeNodeUnary for StreamProjectSet { fn input(&self) -> PlanRef { From 36bbb75d7b191c21a5939e498b4dff7b6eeef862 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 04:56:57 +0000 Subject: [PATCH 16/22] Cargo fix --- src/frontend/src/optimizer/plan_node/batch_expand.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_filter.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_group_topn.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_hash_agg.rs | 2 +- src/frontend/src/optimizer/plan_node/generic/project.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_agg.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_dedup.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_delete.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_filter.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_hop_window.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_over_window.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_project.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_topn.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_update.rs | 2 +- src/frontend/src/optimizer/plan_node/stream_dedup.rs | 2 +- src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs | 2 +- src/frontend/src/optimizer/plan_node/stream_expand.rs | 2 +- src/frontend/src/optimizer/plan_node/stream_filter.rs | 2 +- src/frontend/src/optimizer/plan_node/stream_project_set.rs | 2 +- 19 files changed, 19 insertions(+), 19 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/batch_expand.rs b/src/frontend/src/optimizer/plan_node/batch_expand.rs index de213a30500f..dfa5bf7b7703 100644 --- a/src/frontend/src/optimizer/plan_node/batch_expand.rs +++ b/src/frontend/src/optimizer/plan_node/batch_expand.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/batch_filter.rs b/src/frontend/src/optimizer/plan_node/batch_filter.rs index 052b6b5398ba..dec86bff0007 100644 --- a/src/frontend/src/optimizer/plan_node/batch_filter.rs +++ b/src/frontend/src/optimizer/plan_node/batch_filter.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/batch_group_topn.rs b/src/frontend/src/optimizer/plan_node/batch_group_topn.rs index 889d9385e162..ed8812d255a2 100644 --- a/src/frontend/src/optimizer/plan_node/batch_group_topn.rs +++ b/src/frontend/src/optimizer/plan_node/batch_group_topn.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs b/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs index 7e0b21b8d7ba..66c3f9d878a7 100644 --- a/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/generic/project.rs b/src/frontend/src/optimizer/plan_node/generic/project.rs index 10024aeb4d0c..2b098966d918 100644 --- a/src/frontend/src/optimizer/plan_node/generic/project.rs +++ b/src/frontend/src/optimizer/plan_node/generic/project.rs @@ -21,7 +21,7 @@ use pretty_xmlish::{Pretty, StrAssocArr}; use risingwave_common::catalog::{Field, Schema}; use risingwave_common::util::iter_util::ZipEqFast; -use super::{impl_distill_unit_from_fields, GenericPlanNode, GenericPlanRef}; +use super::{GenericPlanNode, GenericPlanRef}; use crate::expr::{assert_input_ref, Expr, ExprDisplay, ExprImpl, ExprRewriter, InputRef}; use crate::optimizer::optimizer_context::OptimizerContextRef; use crate::optimizer::property::FunctionalDependencySet; diff --git a/src/frontend/src/optimizer/plan_node/logical_agg.rs b/src/frontend/src/optimizer/plan_node/logical_agg.rs index 7759029f8213..dc45a097911e 100644 --- a/src/frontend/src/optimizer/plan_node/logical_agg.rs +++ b/src/frontend/src/optimizer/plan_node/logical_agg.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/logical_dedup.rs b/src/frontend/src/optimizer/plan_node/logical_dedup.rs index daa26174231e..0efcc7c50a6b 100644 --- a/src/frontend/src/optimizer/plan_node/logical_dedup.rs +++ b/src/frontend/src/optimizer/plan_node/logical_dedup.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use core::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/logical_delete.rs b/src/frontend/src/optimizer/plan_node/logical_delete.rs index 2b16e71c793e..0eafcc9bc76c 100644 --- a/src/frontend/src/optimizer/plan_node/logical_delete.rs +++ b/src/frontend/src/optimizer/plan_node/logical_delete.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{fmt, vec}; +use std::{vec}; use risingwave_common::catalog::{Field, Schema, TableVersionId}; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/logical_filter.rs b/src/frontend/src/optimizer/plan_node/logical_filter.rs index cb38fc146dd3..013ee74bcff0 100644 --- a/src/frontend/src/optimizer/plan_node/logical_filter.rs +++ b/src/frontend/src/optimizer/plan_node/logical_filter.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/logical_hop_window.rs b/src/frontend/src/optimizer/plan_node/logical_hop_window.rs index 681ad8ab0522..1fb4200fea28 100644 --- a/src/frontend/src/optimizer/plan_node/logical_hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/logical_hop_window.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/logical_over_window.rs b/src/frontend/src/optimizer/plan_node/logical_over_window.rs index 34fb1ea34446..21dfb4356dbb 100644 --- a/src/frontend/src/optimizer/plan_node/logical_over_window.rs +++ b/src/frontend/src/optimizer/plan_node/logical_over_window.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/logical_project.rs b/src/frontend/src/optimizer/plan_node/logical_project.rs index 6fe21375e251..82883d4ac66c 100644 --- a/src/frontend/src/optimizer/plan_node/logical_project.rs +++ b/src/frontend/src/optimizer/plan_node/logical_project.rs @@ -25,7 +25,7 @@ use super::{ PlanTreeNodeUnary, PredicatePushdown, StreamProject, ToBatch, ToStream, }; use crate::expr::{collect_input_refs, ExprImpl, ExprRewriter, InputRef}; -use crate::optimizer::plan_node::generic::{DistillUnit, GenericPlanRef}; +use crate::optimizer::plan_node::generic::{GenericPlanRef}; use crate::optimizer::plan_node::{ ColumnPruningContext, PredicatePushdownContext, RewriteStreamContext, ToStreamContext, }; diff --git a/src/frontend/src/optimizer/plan_node/logical_topn.rs b/src/frontend/src/optimizer/plan_node/logical_topn.rs index 03707622811e..966926a5acce 100644 --- a/src/frontend/src/optimizer/plan_node/logical_topn.rs +++ b/src/frontend/src/optimizer/plan_node/logical_topn.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/logical_update.rs b/src/frontend/src/optimizer/plan_node/logical_update.rs index 7a10ff500ea8..269985327c64 100644 --- a/src/frontend/src/optimizer/plan_node/logical_update.rs +++ b/src/frontend/src/optimizer/plan_node/logical_update.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{fmt, vec}; +use std::{vec}; use risingwave_common::catalog::{Field, Schema, TableVersionId}; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/stream_dedup.rs b/src/frontend/src/optimizer/plan_node/stream_dedup.rs index e2945b6a1500..07eb5eef3de1 100644 --- a/src/frontend/src/optimizer/plan_node/stream_dedup.rs +++ b/src/frontend/src/optimizer/plan_node/stream_dedup.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use itertools::Itertools; use risingwave_common::util::sort_util::OrderType; diff --git a/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs b/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs index 30816f7dfbfb..ca8bc5a761da 100644 --- a/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs +++ b/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::collections::HashSet; -use std::fmt; + use fixedbitset::FixedBitSet; use risingwave_common::util::sort_util::OrderType; diff --git a/src/frontend/src/optimizer/plan_node/stream_expand.rs b/src/frontend/src/optimizer/plan_node/stream_expand.rs index bca10ed77834..8c9cfa211789 100644 --- a/src/frontend/src/optimizer/plan_node/stream_expand.rs +++ b/src/frontend/src/optimizer/plan_node/stream_expand.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use risingwave_pb::stream_plan::expand_node::Subset; diff --git a/src/frontend/src/optimizer/plan_node/stream_filter.rs b/src/frontend/src/optimizer/plan_node/stream_filter.rs index 93b4b1787795..dcc28d54c611 100644 --- a/src/frontend/src/optimizer/plan_node/stream_filter.rs +++ b/src/frontend/src/optimizer/plan_node/stream_filter.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::FilterNode; diff --git a/src/frontend/src/optimizer/plan_node/stream_project_set.rs b/src/frontend/src/optimizer/plan_node/stream_project_set.rs index a673bb830b36..88950ce144c5 100644 --- a/src/frontend/src/optimizer/plan_node/stream_project_set.rs +++ b/src/frontend/src/optimizer/plan_node/stream_project_set.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; From e596818381e74655d7e8c26badaa485fbce9b3a2 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 05:02:51 +0000 Subject: [PATCH 17/22] Provide more impls --- src/frontend/src/optimizer/plan_node/batch_hop_window.rs | 8 ++------ src/frontend/src/optimizer/plan_node/batch_insert.rs | 7 ++----- src/frontend/src/optimizer/plan_node/batch_project_set.rs | 7 ++----- src/frontend/src/optimizer/plan_node/batch_simple_agg.rs | 8 ++------ src/frontend/src/optimizer/plan_node/batch_sort_agg.rs | 8 ++------ src/frontend/src/optimizer/plan_node/batch_topn.rs | 7 ++----- src/frontend/src/optimizer/plan_node/batch_update.rs | 7 ++----- src/frontend/src/optimizer/plan_node/logical_project.rs | 4 ---- .../optimizer/plan_node/stream_stateless_simple_agg.rs | 8 ++------ 9 files changed, 16 insertions(+), 48 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/batch_hop_window.rs b/src/frontend/src/optimizer/plan_node/batch_hop_window.rs index fab259b0747d..ad2e652f4f3e 100644 --- a/src/frontend/src/optimizer/plan_node/batch_hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/batch_hop_window.rs @@ -18,6 +18,7 @@ use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::HopWindowNode; +use super::utils::impl_distill_by_unit; use super::{ generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -58,12 +59,7 @@ impl BatchHopWindow { } } } - -impl fmt::Display for BatchHopWindow { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchHopWindow") - } -} +impl_distill_by_unit!(BatchHopWindow, logical, "BatchHopWindow"); impl PlanTreeNodeUnary for BatchHopWindow { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_insert.rs b/src/frontend/src/optimizer/plan_node/batch_insert.rs index 95f88c0db19d..4cdcc6b2fbce 100644 --- a/src/frontend/src/optimizer/plan_node/batch_insert.rs +++ b/src/frontend/src/optimizer/plan_node/batch_insert.rs @@ -20,6 +20,7 @@ use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::InsertNode; use risingwave_pb::plan_common::{DefaultColumns, IndexAndExpr}; +use super::utils::impl_distill_by_unit; use super::{ ExprRewritable, LogicalInsert, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -48,11 +49,7 @@ impl BatchInsert { } } -impl fmt::Display for BatchInsert { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchInsert") - } -} +impl_distill_by_unit!(BatchInsert, logical, "BatchInsert"); impl PlanTreeNodeUnary for BatchInsert { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_project_set.rs b/src/frontend/src/optimizer/plan_node/batch_project_set.rs index 765dfab71035..a60a2010195e 100644 --- a/src/frontend/src/optimizer/plan_node/batch_project_set.rs +++ b/src/frontend/src/optimizer/plan_node/batch_project_set.rs @@ -19,6 +19,7 @@ use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::ProjectSetNode; +use super::utils::impl_distill_by_unit; use super::{generic, ExprRewritable}; use crate::expr::ExprRewriter; use crate::optimizer::plan_node::{ @@ -48,11 +49,7 @@ impl BatchProjectSet { } } -impl fmt::Display for BatchProjectSet { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchProjectSet") - } -} +impl_distill_by_unit!(BatchProjectSet, logical, "BatchProjectSet"); impl PlanTreeNodeUnary for BatchProjectSet { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs b/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs index d32a90e9247e..c9f9bd9352b4 100644 --- a/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs @@ -19,6 +19,7 @@ use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::SortAggNode; use super::generic::{self, PlanAggCall}; +use super::utils::impl_distill_by_unit; use super::{ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch}; use crate::expr::ExprRewriter; use crate::optimizer::plan_node::{BatchExchange, ToLocalBatch}; @@ -50,12 +51,7 @@ impl BatchSimpleAgg { self.logical.can_two_phase_agg() && self.two_phase_agg_enabled() } } - -impl fmt::Display for BatchSimpleAgg { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchSimpleAgg") - } -} +impl_distill_by_unit!(BatchSimpleAgg, logical, "BatchSimpleAgg"); impl PlanTreeNodeUnary for BatchSimpleAgg { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs b/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs index 31552750a904..246b7105eefb 100644 --- a/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs @@ -22,6 +22,7 @@ use risingwave_pb::batch_plan::SortAggNode; use risingwave_pb::expr::ExprNode; use super::generic::{self, GenericPlanRef, PlanAggCall}; +use super::utils::impl_distill_by_unit; use super::{ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch}; use crate::expr::{Expr, ExprImpl, ExprRewriter, InputRef}; use crate::optimizer::plan_node::ToLocalBatch; @@ -78,12 +79,7 @@ impl BatchSortAgg { &self.logical.group_key } } - -impl fmt::Display for BatchSortAgg { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchSortAgg") - } -} +impl_distill_by_unit!(BatchSortAgg, logical, "BatchSortAgg"); impl PlanTreeNodeUnary for BatchSortAgg { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_topn.rs b/src/frontend/src/optimizer/plan_node/batch_topn.rs index be5ac6559720..2d0c7f005aa4 100644 --- a/src/frontend/src/optimizer/plan_node/batch_topn.rs +++ b/src/frontend/src/optimizer/plan_node/batch_topn.rs @@ -19,6 +19,7 @@ use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::TopNNode; use super::generic::Limit; +use super::utils::impl_distill_by_unit; use super::{ generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -79,11 +80,7 @@ impl BatchTopN { } } -impl fmt::Display for BatchTopN { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchTopN") - } -} +impl_distill_by_unit!(BatchTopN, logical, "BatchTopN"); impl PlanTreeNodeUnary for BatchTopN { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/batch_update.rs b/src/frontend/src/optimizer/plan_node/batch_update.rs index 5927785144bd..56e7d5d9c7ea 100644 --- a/src/frontend/src/optimizer/plan_node/batch_update.rs +++ b/src/frontend/src/optimizer/plan_node/batch_update.rs @@ -19,6 +19,7 @@ use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::UpdateNode; +use super::utils::impl_distill_by_unit; use super::{ generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -41,11 +42,7 @@ impl BatchUpdate { } } -impl fmt::Display for BatchUpdate { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "BatchUpdate") - } -} +impl_distill_by_unit!(BatchUpdate, logical, "BatchUpdate"); impl PlanTreeNodeUnary for BatchUpdate { fn input(&self) -> PlanRef { diff --git a/src/frontend/src/optimizer/plan_node/logical_project.rs b/src/frontend/src/optimizer/plan_node/logical_project.rs index 82883d4ac66c..3b2bf8b99026 100644 --- a/src/frontend/src/optimizer/plan_node/logical_project.rs +++ b/src/frontend/src/optimizer/plan_node/logical_project.rs @@ -86,10 +86,6 @@ impl LogicalProject { &self.core.exprs } - pub(super) fn fmt_with_name(&self, f: &mut fmt::Formatter<'_>, name: &str) -> fmt::Result { - self.core.fmt_with_name(f, name, self.base.schema()) - } - pub fn is_identity(&self) -> bool { self.core.is_identity() } diff --git a/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs b/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs index 28a251f4949e..9483561822dd 100644 --- a/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs +++ b/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs @@ -19,6 +19,7 @@ use itertools::Itertools; use risingwave_pb::stream_plan::stream_node::PbNodeBody; use super::generic::{self, PlanAggCall}; +use super::utils::impl_distill_by_unit; use super::{ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, StreamNode}; use crate::expr::ExprRewriter; use crate::optimizer::plan_node::stream::StreamPlanRef; @@ -65,12 +66,7 @@ impl StreamStatelessSimpleAgg { &self.logical.agg_calls } } - -impl fmt::Display for StreamStatelessSimpleAgg { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - self.logical.fmt_with_name(f, "StreamStatelessSimpleAgg") - } -} +impl_distill_by_unit!(StreamStatelessSimpleAgg, logical, "StreamStatelessSimpleAgg"); impl PlanTreeNodeUnary for StreamStatelessSimpleAgg { fn input(&self) -> PlanRef { From 0fbc3beaf90301f39d26e28ac8d1fed641dfb3b0 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 05:03:38 +0000 Subject: [PATCH 18/22] Comment out --- src/frontend/src/optimizer/plan_node/batch_insert.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/batch_insert.rs b/src/frontend/src/optimizer/plan_node/batch_insert.rs index 4cdcc6b2fbce..f954dc04c2d8 100644 --- a/src/frontend/src/optimizer/plan_node/batch_insert.rs +++ b/src/frontend/src/optimizer/plan_node/batch_insert.rs @@ -20,7 +20,6 @@ use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::InsertNode; use risingwave_pb::plan_common::{DefaultColumns, IndexAndExpr}; -use super::utils::impl_distill_by_unit; use super::{ ExprRewritable, LogicalInsert, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch, }; @@ -49,7 +48,12 @@ impl BatchInsert { } } -impl_distill_by_unit!(BatchInsert, logical, "BatchInsert"); +impl fmt::Display for BatchInsert { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + self.logical.fmt_with_name(f, "BatchInsert") + } +} +// impl_distill_by_unit!(BatchInsert, logical, "BatchInsert"); impl PlanTreeNodeUnary for BatchInsert { fn input(&self) -> PlanRef { From 76e11c2c21fb3ccf49dd395ffe142784aa8757ed Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 05:04:32 +0000 Subject: [PATCH 19/22] Fmt --- src/frontend/src/optimizer/plan_node/batch_hop_window.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_project_set.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_simple_agg.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_sort_agg.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_topn.rs | 2 +- src/frontend/src/optimizer/plan_node/batch_update.rs | 2 +- .../src/optimizer/plan_node/stream_stateless_simple_agg.rs | 2 +- 7 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/batch_hop_window.rs b/src/frontend/src/optimizer/plan_node/batch_hop_window.rs index ad2e652f4f3e..85d35344a168 100644 --- a/src/frontend/src/optimizer/plan_node/batch_hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/batch_hop_window.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/batch_project_set.rs b/src/frontend/src/optimizer/plan_node/batch_project_set.rs index a60a2010195e..973930767e75 100644 --- a/src/frontend/src/optimizer/plan_node/batch_project_set.rs +++ b/src/frontend/src/optimizer/plan_node/batch_project_set.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs b/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs index c9f9bd9352b4..7bc29db461ca 100644 --- a/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs b/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs index 246b7105eefb..37d214d6f30c 100644 --- a/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; diff --git a/src/frontend/src/optimizer/plan_node/batch_topn.rs b/src/frontend/src/optimizer/plan_node/batch_topn.rs index 2d0c7f005aa4..5f8542c42172 100644 --- a/src/frontend/src/optimizer/plan_node/batch_topn.rs +++ b/src/frontend/src/optimizer/plan_node/batch_topn.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/batch_update.rs b/src/frontend/src/optimizer/plan_node/batch_update.rs index 56e7d5d9c7ea..30d5dae27034 100644 --- a/src/frontend/src/optimizer/plan_node/batch_update.rs +++ b/src/frontend/src/optimizer/plan_node/batch_update.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use risingwave_common::catalog::Schema; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs b/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs index 9483561822dd..deef91cd9427 100644 --- a/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs +++ b/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use fixedbitset::FixedBitSet; use itertools::Itertools; From c0a490bd6bbcd8e67e24956f99ef3f586983e65d Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 05:05:59 +0000 Subject: [PATCH 20/22] Unused --- src/frontend/src/optimizer/plan_node/logical_except.rs | 4 ---- src/frontend/src/optimizer/plan_node/logical_expand.rs | 4 ---- src/frontend/src/optimizer/plan_node/logical_intersect.rs | 4 ---- 3 files changed, 12 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/logical_except.rs b/src/frontend/src/optimizer/plan_node/logical_except.rs index 15bc823c3186..0995e812c10c 100644 --- a/src/frontend/src/optimizer/plan_node/logical_except.rs +++ b/src/frontend/src/optimizer/plan_node/logical_except.rs @@ -47,10 +47,6 @@ impl LogicalExcept { LogicalExcept::new(all, inputs).into() } - pub(super) fn fmt_with_name(&self, f: &mut fmt::Formatter<'_>, name: &str) -> fmt::Result { - self.core.fmt_with_name(f, name) - } - pub fn all(&self) -> bool { self.core.all } diff --git a/src/frontend/src/optimizer/plan_node/logical_expand.rs b/src/frontend/src/optimizer/plan_node/logical_expand.rs index acb2abf89181..745cfd532bfc 100644 --- a/src/frontend/src/optimizer/plan_node/logical_expand.rs +++ b/src/frontend/src/optimizer/plan_node/logical_expand.rs @@ -63,10 +63,6 @@ impl LogicalExpand { pub fn column_subsets(&self) -> &Vec> { &self.core.column_subsets } - - pub(super) fn fmt_with_name(&self, f: &mut fmt::Formatter<'_>, name: &str) -> fmt::Result { - self.core.fmt_with_name(f, name) - } } impl PlanTreeNodeUnary for LogicalExpand { diff --git a/src/frontend/src/optimizer/plan_node/logical_intersect.rs b/src/frontend/src/optimizer/plan_node/logical_intersect.rs index 6c92edb0bbcf..10013e2e5794 100644 --- a/src/frontend/src/optimizer/plan_node/logical_intersect.rs +++ b/src/frontend/src/optimizer/plan_node/logical_intersect.rs @@ -46,10 +46,6 @@ impl LogicalIntersect { LogicalIntersect::new(all, inputs).into() } - pub(super) fn fmt_with_name(&self, f: &mut fmt::Formatter<'_>, name: &str) -> fmt::Result { - self.core.fmt_with_name(f, name) - } - pub fn fmt_fields_with_builder(&self, builder: &mut fmt::DebugStruct<'_, '_>) { self.core.fmt_fields_with_builder(builder) } From 488058af47c355e41368d1cb0ff55489998d50b0 Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 05:06:49 +0000 Subject: [PATCH 21/22] Fixed --- src/frontend/src/optimizer/plan_node/logical_except.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_expand.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/logical_except.rs b/src/frontend/src/optimizer/plan_node/logical_except.rs index 0995e812c10c..2feaefea246e 100644 --- a/src/frontend/src/optimizer/plan_node/logical_except.rs +++ b/src/frontend/src/optimizer/plan_node/logical_except.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use itertools::Itertools; use risingwave_common::catalog::Schema; diff --git a/src/frontend/src/optimizer/plan_node/logical_expand.rs b/src/frontend/src/optimizer/plan_node/logical_expand.rs index 745cfd532bfc..d832b18a28e4 100644 --- a/src/frontend/src/optimizer/plan_node/logical_expand.rs +++ b/src/frontend/src/optimizer/plan_node/logical_expand.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::fmt; + use itertools::Itertools; use risingwave_common::error::Result; From f2bbfc6d8af66e951547640c8b295559071fd70b Mon Sep 17 00:00:00 2001 From: ice1000 Date: Mon, 5 Jun 2023 05:06:58 +0000 Subject: [PATCH 22/22] Fmt --- src/frontend/src/optimizer/plan_node/batch_expand.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_filter.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_group_topn.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_hash_agg.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_hop_window.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_project_set.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_simple_agg.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_sort_agg.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_topn.rs | 2 -- src/frontend/src/optimizer/plan_node/batch_update.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_agg.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_dedup.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_delete.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_except.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_expand.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_filter.rs | 2 -- .../src/optimizer/plan_node/logical_hop_window.rs | 2 -- .../src/optimizer/plan_node/logical_over_window.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_project.rs | 2 +- src/frontend/src/optimizer/plan_node/logical_topn.rs | 2 -- src/frontend/src/optimizer/plan_node/logical_update.rs | 2 +- src/frontend/src/optimizer/plan_node/stream_dedup.rs | 4 +--- .../src/optimizer/plan_node/stream_eowc_over_window.rs | 3 +-- src/frontend/src/optimizer/plan_node/stream_expand.rs | 2 -- src/frontend/src/optimizer/plan_node/stream_filter.rs | 2 -- .../src/optimizer/plan_node/stream_project_set.rs | 2 -- .../optimizer/plan_node/stream_stateless_simple_agg.rs | 8 +++++--- 27 files changed, 10 insertions(+), 53 deletions(-) diff --git a/src/frontend/src/optimizer/plan_node/batch_expand.rs b/src/frontend/src/optimizer/plan_node/batch_expand.rs index dfa5bf7b7703..72caa27858be 100644 --- a/src/frontend/src/optimizer/plan_node/batch_expand.rs +++ b/src/frontend/src/optimizer/plan_node/batch_expand.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use itertools::Itertools; use risingwave_common::error::Result; use risingwave_pb::batch_plan::expand_node::Subset; diff --git a/src/frontend/src/optimizer/plan_node/batch_filter.rs b/src/frontend/src/optimizer/plan_node/batch_filter.rs index dec86bff0007..aadbda9800b1 100644 --- a/src/frontend/src/optimizer/plan_node/batch_filter.rs +++ b/src/frontend/src/optimizer/plan_node/batch_filter.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::FilterNode; diff --git a/src/frontend/src/optimizer/plan_node/batch_group_topn.rs b/src/frontend/src/optimizer/plan_node/batch_group_topn.rs index ed8812d255a2..1d61b4e9eb37 100644 --- a/src/frontend/src/optimizer/plan_node/batch_group_topn.rs +++ b/src/frontend/src/optimizer/plan_node/batch_group_topn.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::GroupTopNNode; diff --git a/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs b/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs index 66c3f9d878a7..a05d8ff915e7 100644 --- a/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_hash_agg.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/batch_hop_window.rs b/src/frontend/src/optimizer/plan_node/batch_hop_window.rs index 85d35344a168..c4b84b7232d1 100644 --- a/src/frontend/src/optimizer/plan_node/batch_hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/batch_hop_window.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::HopWindowNode; diff --git a/src/frontend/src/optimizer/plan_node/batch_project_set.rs b/src/frontend/src/optimizer/plan_node/batch_project_set.rs index 973930767e75..b86211aaaa21 100644 --- a/src/frontend/src/optimizer/plan_node/batch_project_set.rs +++ b/src/frontend/src/optimizer/plan_node/batch_project_set.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use itertools::Itertools; use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs b/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs index 7bc29db461ca..b9c747c20505 100644 --- a/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_simple_agg.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::SortAggNode; diff --git a/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs b/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs index 37d214d6f30c..c81bbc7bbde8 100644 --- a/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs +++ b/src/frontend/src/optimizer/plan_node/batch_sort_agg.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/batch_topn.rs b/src/frontend/src/optimizer/plan_node/batch_topn.rs index 5f8542c42172..f6f61410889f 100644 --- a/src/frontend/src/optimizer/plan_node/batch_topn.rs +++ b/src/frontend/src/optimizer/plan_node/batch_topn.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; use risingwave_pb::batch_plan::TopNNode; diff --git a/src/frontend/src/optimizer/plan_node/batch_update.rs b/src/frontend/src/optimizer/plan_node/batch_update.rs index 30d5dae27034..15ab0ca59ba7 100644 --- a/src/frontend/src/optimizer/plan_node/batch_update.rs +++ b/src/frontend/src/optimizer/plan_node/batch_update.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_common::catalog::Schema; use risingwave_common::error::Result; use risingwave_pb::batch_plan::plan_node::NodeBody; diff --git a/src/frontend/src/optimizer/plan_node/logical_agg.rs b/src/frontend/src/optimizer/plan_node/logical_agg.rs index dc45a097911e..bdb587cc2292 100644 --- a/src/frontend/src/optimizer/plan_node/logical_agg.rs +++ b/src/frontend/src/optimizer/plan_node/logical_agg.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::{ErrorCode, Result, TrackingIssue}; diff --git a/src/frontend/src/optimizer/plan_node/logical_dedup.rs b/src/frontend/src/optimizer/plan_node/logical_dedup.rs index 0efcc7c50a6b..6be7020aa57f 100644 --- a/src/frontend/src/optimizer/plan_node/logical_dedup.rs +++ b/src/frontend/src/optimizer/plan_node/logical_dedup.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/logical_delete.rs b/src/frontend/src/optimizer/plan_node/logical_delete.rs index 0eafcc9bc76c..e5b34c6cf4cd 100644 --- a/src/frontend/src/optimizer/plan_node/logical_delete.rs +++ b/src/frontend/src/optimizer/plan_node/logical_delete.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{vec}; +use std::vec; use risingwave_common::catalog::{Field, Schema, TableVersionId}; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/logical_except.rs b/src/frontend/src/optimizer/plan_node/logical_except.rs index 2feaefea246e..11ff8b0210ee 100644 --- a/src/frontend/src/optimizer/plan_node/logical_except.rs +++ b/src/frontend/src/optimizer/plan_node/logical_except.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use itertools::Itertools; use risingwave_common::catalog::Schema; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/logical_expand.rs b/src/frontend/src/optimizer/plan_node/logical_expand.rs index d832b18a28e4..2cd662cf2da6 100644 --- a/src/frontend/src/optimizer/plan_node/logical_expand.rs +++ b/src/frontend/src/optimizer/plan_node/logical_expand.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/logical_filter.rs b/src/frontend/src/optimizer/plan_node/logical_filter.rs index 013ee74bcff0..4ba172aea9b7 100644 --- a/src/frontend/src/optimizer/plan_node/logical_filter.rs +++ b/src/frontend/src/optimizer/plan_node/logical_filter.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::bail; diff --git a/src/frontend/src/optimizer/plan_node/logical_hop_window.rs b/src/frontend/src/optimizer/plan_node/logical_hop_window.rs index 1fb4200fea28..530b4a198875 100644 --- a/src/frontend/src/optimizer/plan_node/logical_hop_window.rs +++ b/src/frontend/src/optimizer/plan_node/logical_hop_window.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/logical_over_window.rs b/src/frontend/src/optimizer/plan_node/logical_over_window.rs index 21dfb4356dbb..d24cf86541bf 100644 --- a/src/frontend/src/optimizer/plan_node/logical_over_window.rs +++ b/src/frontend/src/optimizer/plan_node/logical_over_window.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::{ErrorCode, Result}; diff --git a/src/frontend/src/optimizer/plan_node/logical_project.rs b/src/frontend/src/optimizer/plan_node/logical_project.rs index 3b2bf8b99026..b8a3ef8a2e32 100644 --- a/src/frontend/src/optimizer/plan_node/logical_project.rs +++ b/src/frontend/src/optimizer/plan_node/logical_project.rs @@ -25,7 +25,7 @@ use super::{ PlanTreeNodeUnary, PredicatePushdown, StreamProject, ToBatch, ToStream, }; use crate::expr::{collect_input_refs, ExprImpl, ExprRewriter, InputRef}; -use crate::optimizer::plan_node::generic::{GenericPlanRef}; +use crate::optimizer::plan_node::generic::GenericPlanRef; use crate::optimizer::plan_node::{ ColumnPruningContext, PredicatePushdownContext, RewriteStreamContext, ToStreamContext, }; diff --git a/src/frontend/src/optimizer/plan_node/logical_topn.rs b/src/frontend/src/optimizer/plan_node/logical_topn.rs index 966926a5acce..f09e405b43ca 100644 --- a/src/frontend/src/optimizer/plan_node/logical_topn.rs +++ b/src/frontend/src/optimizer/plan_node/logical_topn.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_common::error::{ErrorCode, Result, RwError}; diff --git a/src/frontend/src/optimizer/plan_node/logical_update.rs b/src/frontend/src/optimizer/plan_node/logical_update.rs index 269985327c64..abe3e6d64849 100644 --- a/src/frontend/src/optimizer/plan_node/logical_update.rs +++ b/src/frontend/src/optimizer/plan_node/logical_update.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{vec}; +use std::vec; use risingwave_common::catalog::{Field, Schema, TableVersionId}; use risingwave_common::error::Result; diff --git a/src/frontend/src/optimizer/plan_node/stream_dedup.rs b/src/frontend/src/optimizer/plan_node/stream_dedup.rs index 07eb5eef3de1..44acf722eae6 100644 --- a/src/frontend/src/optimizer/plan_node/stream_dedup.rs +++ b/src/frontend/src/optimizer/plan_node/stream_dedup.rs @@ -12,15 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use itertools::Itertools; use risingwave_common::util::sort_util::OrderType; use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::DedupNode; use super::generic::{self, GenericPlanNode, GenericPlanRef}; -use super::utils::{TableCatalogBuilder, impl_distill_by_unit}; +use super::utils::{impl_distill_by_unit, TableCatalogBuilder}; use super::{ExprRewritable, PlanBase, PlanTreeNodeUnary, StreamNode}; use crate::optimizer::plan_node::stream::StreamPlanRef; use crate::optimizer::plan_node::PlanRef; diff --git a/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs b/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs index ca8bc5a761da..dea3d0eb4988 100644 --- a/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs +++ b/src/frontend/src/optimizer/plan_node/stream_eowc_over_window.rs @@ -14,13 +14,12 @@ use std::collections::HashSet; - use fixedbitset::FixedBitSet; use risingwave_common::util::sort_util::OrderType; use risingwave_pb::stream_plan::stream_node::PbNodeBody; use super::generic::{self, PlanWindowFunction}; -use super::utils::{TableCatalogBuilder, impl_distill_by_unit}; +use super::utils::{impl_distill_by_unit, TableCatalogBuilder}; use super::{ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, StreamNode}; use crate::optimizer::plan_node::stream::StreamPlanRef; use crate::stream_fragmenter::BuildFragmentGraphState; diff --git a/src/frontend/src/optimizer/plan_node/stream_expand.rs b/src/frontend/src/optimizer/plan_node/stream_expand.rs index 8c9cfa211789..88b5ae12df4d 100644 --- a/src/frontend/src/optimizer/plan_node/stream_expand.rs +++ b/src/frontend/src/optimizer/plan_node/stream_expand.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use risingwave_pb::stream_plan::expand_node::Subset; use risingwave_pb::stream_plan::stream_node::PbNodeBody; diff --git a/src/frontend/src/optimizer/plan_node/stream_filter.rs b/src/frontend/src/optimizer/plan_node/stream_filter.rs index dcc28d54c611..760ac76d1e87 100644 --- a/src/frontend/src/optimizer/plan_node/stream_filter.rs +++ b/src/frontend/src/optimizer/plan_node/stream_filter.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use risingwave_pb::stream_plan::stream_node::PbNodeBody; use risingwave_pb::stream_plan::FilterNode; diff --git a/src/frontend/src/optimizer/plan_node/stream_project_set.rs b/src/frontend/src/optimizer/plan_node/stream_project_set.rs index 88950ce144c5..566bbde6675e 100644 --- a/src/frontend/src/optimizer/plan_node/stream_project_set.rs +++ b/src/frontend/src/optimizer/plan_node/stream_project_set.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_pb::stream_plan::stream_node::PbNodeBody; diff --git a/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs b/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs index deef91cd9427..546687130632 100644 --- a/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs +++ b/src/frontend/src/optimizer/plan_node/stream_stateless_simple_agg.rs @@ -12,8 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. - - use fixedbitset::FixedBitSet; use itertools::Itertools; use risingwave_pb::stream_plan::stream_node::PbNodeBody; @@ -66,7 +64,11 @@ impl StreamStatelessSimpleAgg { &self.logical.agg_calls } } -impl_distill_by_unit!(StreamStatelessSimpleAgg, logical, "StreamStatelessSimpleAgg"); +impl_distill_by_unit!( + StreamStatelessSimpleAgg, + logical, + "StreamStatelessSimpleAgg" +); impl PlanTreeNodeUnary for StreamStatelessSimpleAgg { fn input(&self) -> PlanRef {