refactor(plan_node_fmt): further prepare generic code where they use fmt_with_name

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 {

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 {

src/frontend/src/optimizer/plan_node/batch_expand.rs

@@ -12,14 +12,13 @@
 // 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;
 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 +50,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 {

src/frontend/src/optimizer/plan_node/batch_filter.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::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 +44,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 {

src/frontend/src/optimizer/plan_node/batch_group_topn.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::GroupTopNNode;
 
+use super::utils::impl_distill_by_unit;
 use super::{
     generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch,
 };
@@ -47,11 +46,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 {

src/frontend/src/optimizer/plan_node/batch_hash_agg.rs

@@ -12,15 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::fmt;
-
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use risingwave_common::error::Result;
 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 +99,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 {

src/frontend/src/optimizer/plan_node/batch_hop_window.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::HopWindowNode;
 
+use super::utils::impl_distill_by_unit;
 use super::{
     generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch,
 };
@@ -58,12 +57,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 {

src/frontend/src/optimizer/plan_node/batch_insert.rs

@@ -53,6 +53,7 @@ impl fmt::Display for BatchInsert {
         self.logical.fmt_with_name(f, "BatchInsert")
     }
 }
+// impl_distill_by_unit!(BatchInsert, logical, "BatchInsert");
 
 impl PlanTreeNodeUnary for BatchInsert {
     fn input(&self) -> PlanRef {

src/frontend/src/optimizer/plan_node/batch_project_set.rs

@@ -12,13 +12,12 @@
 // 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;
 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 +47,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 {

src/frontend/src/optimizer/plan_node/batch_simple_agg.rs

@@ -12,13 +12,12 @@
 // 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::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 +49,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 {

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 std::fmt;
-
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use risingwave_common::error::Result;
@@ -22,6 +20,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 +77,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 {

src/frontend/src/optimizer/plan_node/batch_topn.rs

@@ -12,13 +12,12 @@
 // 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::TopNNode;
 
 use super::generic::Limit;
+use super::utils::impl_distill_by_unit;
 use super::{
     generic, ExprRewritable, PlanBase, PlanRef, PlanTreeNodeUnary, ToBatchPb, ToDistributedBatch,
 };
@@ -79,11 +78,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 {

src/frontend/src/optimizer/plan_node/batch_update.rs

@@ -12,13 +12,12 @@
 // 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;
 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 +40,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 {