Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/src/frontend/src/handler/explain.rs b/src/frontend/src/handler/explain.rs
index 018b03feebe50..20a7036dc1487 100644
--- a/src/frontend/src/handler/explain.rs
+++ b/src/frontend/src/handler/explain.rs
@@ -232,6 +232,7 @@ async fn do_handle_explain(
                         ExplainFormat::Json => blocks.push(plan.explain_to_json()),
                         ExplainFormat::Xml => blocks.push(plan.explain_to_xml()),
                         ExplainFormat::Yaml => blocks.push(plan.explain_to_yaml()),
+                        ExplainFormat::Dot => blocks.push(plan.explain_to_dot()),
                     }
                 }
             }
diff --git a/src/frontend/src/optimizer/logical_optimization.rs b/src/frontend/src/optimizer/logical_optimization.rs
index bdb75728ff9e8..4116ba77c2190 100644
--- a/src/frontend/src/optimizer/logical_optimization.rs
+++ b/src/frontend/src/optimizer/logical_optimization.rs
@@ -700,6 +700,9 @@ impl LogicalOptimizer {
                 ExplainFormat::Yaml => {
                     ctx.store_logical(plan.explain_to_yaml());
                 }
+                ExplainFormat::Dot => {
+                    ctx.store_logical(plan.explain_to_dot());
+                }
             }
         }
 
@@ -819,6 +822,9 @@ impl LogicalOptimizer {
                 ExplainFormat::Yaml => {
                     ctx.store_logical(plan.explain_to_yaml());
                 }
+                ExplainFormat::Dot => {
+                    ctx.store_logical(plan.explain_to_dot());
+                }
             }
         }
 
diff --git a/src/frontend/src/optimizer/plan_node/mod.rs b/src/frontend/src/optimizer/plan_node/mod.rs
index 9986b8800b66f..165f867b3c76e 100644
--- a/src/frontend/src/optimizer/plan_node/mod.rs
+++ b/src/frontend/src/optimizer/plan_node/mod.rs
@@ -27,6 +27,7 @@
 //! - all field should be valued in construction, so the properties' derivation should be finished
 //!   in the `new()` function.
 
+use std::collections::HashMap;
 use std::fmt::Debug;
 use std::hash::Hash;
 use std::ops::Deref;
@@ -37,6 +38,8 @@ use dyn_clone::DynClone;
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use paste::paste;
+use petgraph::dot::{Config, Dot};
+use petgraph::graph::{Graph, NodeIndex};
 use pretty_xmlish::{Pretty, PrettyConfig};
 use risingwave_common::catalog::Schema;
 use risingwave_common::util::recursive::{self, Recurse};
@@ -642,6 +645,9 @@ pub trait Explain {
     /// Write explain the whole plan tree.
     fn explain<'a>(&self) -> Pretty<'a>;
 
+    /// Write explain the whole plan tree with node id.
+    fn explain_with_id<'a>(&self) -> Pretty<'a>;
+
     /// Explain the plan node and return a string.
     fn explain_to_string(&self) -> String;
 
@@ -653,6 +659,9 @@ pub trait Explain {
 
     /// Explain the plan node and return a yaml string.
     fn explain_to_yaml(&self) -> String;
+
+    /// Explain the plan node and return a dot format string.
+    fn explain_to_dot(&self) -> String;
 }
 
 impl Explain for PlanRef {
@@ -666,6 +675,21 @@ impl Explain for PlanRef {
         Pretty::Record(node)
     }
 
+    /// Write explain the whole plan tree with node id.
+    fn explain_with_id<'a>(&self) -> Pretty<'a> {
+        let node_id = self.id();
+        let mut node = self.distill();
+        // NOTE(kwannoel): Can lead to poor performance if plan is very large,
+        // but we want to show the id first.
+        node.fields
+            .insert(0, ("id".into(), Pretty::display(&node_id.0)));
+        let inputs = self.inputs();
+        for input in inputs.iter().peekable() {
+            node.children.push(input.explain_with_id());
+        }
+        Pretty::Record(node)
+    }
+
     /// Explain the plan node and return a string.
     fn explain_to_string(&self) -> String {
         let plan = reorganize_elements_id(self.clone());
@@ -680,7 +704,7 @@ impl Explain for PlanRef {
     fn explain_to_json(&self) -> String {
         let plan = reorganize_elements_id(self.clone());
         let explain_ir = plan.explain();
-        serde_json::to_string_pretty(&PrettySerde(explain_ir))
+        serde_json::to_string_pretty(&PrettySerde(explain_ir, true))
             .expect("failed to serialize plan to json")
     }
 
@@ -688,14 +712,66 @@ impl Explain for PlanRef {
     fn explain_to_xml(&self) -> String {
         let plan = reorganize_elements_id(self.clone());
         let explain_ir = plan.explain();
-        quick_xml::se::to_string(&PrettySerde(explain_ir)).expect("failed to serialize plan to xml")
+        quick_xml::se::to_string(&PrettySerde(explain_ir, true))
+            .expect("failed to serialize plan to xml")
     }
 
     /// Explain the plan node and return a yaml string.
     fn explain_to_yaml(&self) -> String {
         let plan = reorganize_elements_id(self.clone());
         let explain_ir = plan.explain();
-        serde_yaml::to_string(&PrettySerde(explain_ir)).expect("failed to serialize plan to yaml")
+        serde_yaml::to_string(&PrettySerde(explain_ir, true))
+            .expect("failed to serialize plan to yaml")
+    }
+
+    /// Explain the plan node and return a dot format string.
+    fn explain_to_dot(&self) -> String {
+        let plan = reorganize_elements_id(self.clone());
+        let explain_ir = plan.explain_with_id();
+        let mut graph = Graph::<String, String>::new();
+        let mut nodes = HashMap::new();
+        build_graph_from_pretty(&explain_ir, &mut graph, &mut nodes, None);
+        let dot = Dot::with_config(&graph, &[Config::EdgeNoLabel]);
+        dot.to_string()
+    }
+}
+
+fn build_graph_from_pretty(
+    pretty: &Pretty<'_>,
+    graph: &mut Graph<String, String>,
+    nodes: &mut HashMap<String, NodeIndex>,
+    parent_label: Option<&str>,
+) {
+    if let Pretty::Record(r) = pretty {
+        let mut label = String::new();
+        label.push_str(&r.name);
+        for (k, v) in &r.fields {
+            label.push('\n');
+            label.push_str(k);
+            label.push_str(": ");
+            label.push_str(
+                &serde_json::to_string(&PrettySerde(v.clone(), false))
+                    .expect("failed to serialize plan to dot"),
+            );
+        }
+        // output alignment.
+        if !r.fields.is_empty() {
+            label.push('\n');
+        }
+
+        let current_node = *nodes
+            .entry(label.clone())
+            .or_insert_with(|| graph.add_node(label.clone()));
+
+        if let Some(parent_label) = parent_label {
+            if let Some(&parent_node) = nodes.get(parent_label) {
+                graph.add_edge(parent_node, current_node, "contains".to_string());
+            }
+        }
+
+        for child in &r.children {
+            build_graph_from_pretty(child, graph, nodes, Some(&label));
+        }
     }
 }
 
diff --git a/src/frontend/src/utils/pretty_serde.rs b/src/frontend/src/utils/pretty_serde.rs
index 705267c3163b6..e92bb37267a1c 100644
--- a/src/frontend/src/utils/pretty_serde.rs
+++ b/src/frontend/src/utils/pretty_serde.rs
@@ -28,7 +28,9 @@ use pretty_xmlish::Pretty;
 use serde::ser::{SerializeSeq, SerializeStruct};
 use serde::{Serialize, Serializer};
 
-pub struct PrettySerde<'a>(pub Pretty<'a>);
+// Second anymous field is include_children.
+// If true the children information will be serialized.
+pub struct PrettySerde<'a>(pub Pretty<'a>, pub bool);
 
 impl Serialize for PrettySerde<'_> {
     fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
@@ -46,31 +48,33 @@ impl Serialize for PrettySerde<'_> {
                     &node
                         .fields
                         .iter()
-                        .map(|(k, v)| (k.as_ref(), PrettySerde(v.clone())))
+                        .map(|(k, v)| (k.as_ref(), PrettySerde(v.clone(), self.1)))
                         .collect::<BTreeMap<_, _>>(),
                 )?;
-                state.serialize_field(
-                    "children",
-                    &node
-                        .children
-                        .iter()
-                        .map(|c| PrettySerde(c.clone()))
-                        .collect::<Vec<_>>(),
-                )?;
+                if self.1 {
+                    state.serialize_field(
+                        "children",
+                        &node
+                            .children
+                            .iter()
+                            .map(|c| PrettySerde(c.clone(), self.1))
+                            .collect::<Vec<_>>(),
+                    )?;
+                }
                 state.end()
             }
 
             Pretty::Array(elements) => {
                 let mut seq = serializer.serialize_seq(Some(elements.len()))?;
                 for element in elements {
-                    seq.serialize_element(&PrettySerde((*element).clone()))?;
+                    seq.serialize_element(&PrettySerde((*element).clone(), self.1))?;
                 }
                 seq.end()
             }
 
             Pretty::Linearized(inner, size) => {
                 let mut state = serializer.serialize_struct("Linearized", 2)?;
-                state.serialize_field("inner", &PrettySerde((**inner).clone()))?;
+                state.serialize_field("inner", &PrettySerde((**inner).clone(), self.1))?;
                 state.serialize_field("size", size)?;
                 state.end()
             }
@@ -94,7 +98,7 @@ mod tests {
     #[test]
     fn test_pretty_serde() {
         let pretty = Pretty::childless_record("root", vec![("a", Pretty::Text("1".into()))]);
-        let pretty_serde = PrettySerde(pretty);
+        let pretty_serde = PrettySerde(pretty, true);
         let serialized = serde_json::to_string(&pretty_serde).unwrap();
         check(
             serialized,
diff --git a/src/sqlparser/src/ast/mod.rs b/src/sqlparser/src/ast/mod.rs
index 3460465c2a6a9..be9d4ae489f53 100644
--- a/src/sqlparser/src/ast/mod.rs
+++ b/src/sqlparser/src/ast/mod.rs
@@ -1141,6 +1141,7 @@ pub enum ExplainFormat {
     Json,
     Xml,
     Yaml,
+    Dot,
 }
 
 impl fmt::Display for ExplainFormat {
@@ -1150,6 +1151,7 @@ impl fmt::Display for ExplainFormat {
             ExplainFormat::Json => f.write_str("JSON"),
             ExplainFormat::Xml => f.write_str("XML"),
             ExplainFormat::Yaml => f.write_str("YAML"),
+            ExplainFormat::Dot => f.write_str("DOT"),
         }
     }
 }
diff --git a/src/sqlparser/src/keywords.rs b/src/sqlparser/src/keywords.rs
index d3b196d3bcb06..ccc0ef23502bd 100644
--- a/src/sqlparser/src/keywords.rs
+++ b/src/sqlparser/src/keywords.rs
@@ -202,6 +202,7 @@ define_keywords!(
     DISTRIBUTED,
     DISTSQL,
     DO,
+    DOT,
     DOUBLE,
     DROP,
     DYNAMIC,
diff --git a/src/sqlparser/src/parser.rs b/src/sqlparser/src/parser.rs
index f8d449a253c30..cf7d8e4394574 100644
--- a/src/sqlparser/src/parser.rs
+++ b/src/sqlparser/src/parser.rs
@@ -4039,11 +4039,13 @@ impl Parser<'_> {
                             Keyword::JSON,
                             Keyword::XML,
                             Keyword::YAML,
+                            Keyword::DOT,
                         ])? {
                             Keyword::TEXT => ExplainFormat::Text,
                             Keyword::JSON => ExplainFormat::Json,
                             Keyword::XML => ExplainFormat::Xml,
                             Keyword::YAML => ExplainFormat::Yaml,
+                            Keyword::DOT => ExplainFormat::Dot,
                             _ => unreachable!("{}", keyword),
                         }
                     }


Test patch:
diff --git a/src/frontend/planner_test/tests/testdata/input/explain_dot_format.yaml b/src/frontend/planner_test/tests/testdata/input/explain_dot_format.yaml
new file mode 100644
index 0000000000000..377f395b73b8d
--- /dev/null
+++ b/src/frontend/planner_test/tests/testdata/input/explain_dot_format.yaml
@@ -0,0 +1,44 @@
+- name: test dot output format (logical)
+  sql: |
+    CREATE TABLE t (v1 int);
+    explain (logical, format dot) SELECT approx_percentile(0.5) WITHIN GROUP (order by v1) from t;
+  expected_outputs:
+    - explain_output
+- name: test dot output format (batch)
+  sql: |
+    CREATE TABLE t (v1 int);
+    explain (physical, format dot) SELECT approx_percentile(0.5) WITHIN GROUP (order by v1) from t;
+  expected_outputs:
+    - explain_output
+- name: test dot output format (stream)
+  sql: |
+    CREATE TABLE t (v1 int);
+    explain (physical, format dot) create materialized view m1 as SELECT approx_percentile(0.5) WITHIN GROUP (order by v1) from t;
+  expected_outputs:
+    - explain_output
+- name: test long dot output format (stream)
+  sql: |
+    create table t1(a int, b int);
+    create table t2(c int primary key, d int);
+    explain (physical, format dot) create materialized view m1 as SELECT
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col1,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col2,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col3,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col4,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col5,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col6,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col7,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col8,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col9,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col10,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col11,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col12,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col13,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col14,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col15,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col16,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col17,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col18
+    from t1;
+  expected_outputs:
+    - explain_output
\ No newline at end of file
diff --git a/src/frontend/planner_test/tests/testdata/output/explain_dot_format.yaml b/src/frontend/planner_test/tests/testdata/output/explain_dot_format.yaml
new file mode 100644
index 0000000000000..2fce651976e82
--- /dev/null
+++ b/src/frontend/planner_test/tests/testdata/output/explain_dot_format.yaml
@@ -0,0 +1,439 @@
+# This file is automatically generated. See `src/frontend/planner_test/README.md` for more information.
+- name: test dot output format (logical)
+  sql: |
+    CREATE TABLE t (v1 int);
+    explain (logical, format dot) SELECT approx_percentile(0.5) WITHIN GROUP (order by v1) from t;
+  explain_output: |
+    digraph {
+        0 [ label = "LogicalAgg\lid: \"2\"\laggs: [\"approx_percentile($expr1)\"]\l" ]
+        1 [ label = "LogicalProject\lid: \"1\"\lexprs: [\"t.v1::Float64 as $expr1\"]\l" ]
+        2 [ label = "LogicalScan\lid: \"10017\"\ltable: \"t\"\lcolumns: [\"v1\"]\l" ]
+        0 -> 1 [ ]
+        1 -> 2 [ ]
+    }
+- name: test dot output format (batch)
+  sql: |
+    CREATE TABLE t (v1 int);
+    explain (physical, format dot) SELECT approx_percentile(0.5) WITHIN GROUP (order by v1) from t;
+  explain_output: |
+    digraph {
+        0 [ label = "BatchSimpleAgg\lid: \"3\"\laggs: [\"approx_percentile($expr1)\"]\l" ]
+        1 [ label = "BatchExchange\lid: \"2\"\lorder: []\ldist: \"Single\"\l" ]
+        2 [ label = "BatchProject\lid: \"1\"\lexprs: [\"t.v1::Float64 as $expr1\"]\l" ]
+        3 [ label = "BatchScan\lid: \"10033\"\ltable: \"t\"\lcolumns: [\"v1\"]\l" ]
+        0 -> 1 [ ]
+        1 -> 2 [ ]
+        2 -> 3 [ ]
+    }
+- name: test dot output format (stream)
+  sql: |
+    CREATE TABLE t (v1 int);
+    explain (physical, format dot) create materialized view m1 as SELECT approx_percentile(0.5) WITHIN GROUP (order by v1) from t;
+  explain_output: |
+    digraph {
+        0 [ label = "StreamMaterialize\lid: \"3\"\lcolumns: [\"approx_percentile\"]\lstream_key: []\lpk_columns: []\lpk_conflict: \"NoCheck\"\l" ]
+        1 [ label = "StreamGlobalApproxPercentile\lid: \"10046\"\lquantile: \"0.5:Float64\"\lrelative_error: \"0.01:Float64\"\l" ]
+        2 [ label = "StreamExchange\lid: \"2\"\ldist: \"Single\"\l" ]
+        3 [ label = "StreamLocalApproxPercentile\lid: \"10044\"\lpercentile_col: \"$expr1\"\lquantile: \"0.5:Float64\"\lrelative_error: \"0.01:Float64\"\l" ]
+        4 [ label = "StreamProject\lid: \"1\"\lexprs: [\"t.v1::Float64 as $expr1\",\"t._row_id\"]\l" ]
+        5 [ label = "StreamTableScan\lid: \"10049\"\ltable: \"t\"\lcolumns: [\"v1\",\"_row_id\"]\l" ]
+        0 -> 1 [ ]
+        1 -> 2 [ ]
+        2 -> 3 [ ]
+        3 -> 4 [ ]
+        4 -> 5 [ ]
+    }
+- name: test long dot output format (stream)
+  sql: |
+    create table t1(a int, b int);
+    create table t2(c int primary key, d int);
+    explain (physical, format dot) create materialized view m1 as SELECT
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col1,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col2,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col3,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col4,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col5,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col6,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col7,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col8,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col9,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col10,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col11,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col12,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col13,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col14,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col15,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col16,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col17,
+    COALESCE((SELECT b FROM t2 WHERE t1.a = t2.c), 0) col18
+    from t1;
+  explain_output: |
+    digraph {
+        0 [ label = "StreamMaterialize\lid: \"147\"\lcolumns: [\"col1\",\"col2\",\"col3\",\"col4\",\"col5\",\"col6\",\"col7\",\"col8\",\"col9\",\"col10\",\"col11\",\"col12\",\"col13\",\"col14\",\"col15\",\"col16\",\"col17\",\"col18\",\"t1._row_id(hidden)\",\"t1.b(hidden)\",\"t1.a(hidden)\",\"t1.b#1(hidden)\",\"t1.b#2(hidden)\",\"t1.b#3(hidden)\",\"t1.b#4(hidden)\",\"t1.b#5(hidden)\",\"t1.b#6(hidden)\",\"t1.b#7(hidden)\",\"t1.b#8(hidden)\",\"t1.b#9(hidden)\",\"t1.b#10(hidden)\",\"t1.b#11(hidden)\",\"t1.b#12(hidden)\",\"t1.b#13(hidden)\",\"t1.b#14(hidden)\",\"t1.b#15(hidden)\",\"t1.b#16(hidden)\",\"t1.b#17(hidden)\",\"t1.b#18(hidden)\"]\lstream_key: [\"t1._row_id\",\"t1.b\",\"t1.a\",\"t1.b#1\",\"t1.b#2\",\"t1.b#3\",\"t1.b#4\",\"t1.b#5\",\"t1.b#6\",\"t1.b#7\",\"t1.b#8\",\"t1.b#9\",\"t1.b#10\",\"t1.b#11\",\"t1.b#12\",\"t1.b#13\",\"t1.b#14\",\"t1.b#15\",\"t1.b#16\",\"t1.b#17\",\"t1.b#18\"]\lpk_columns: [\"t1._row_id\",\"t1.b\",\"t1.a\",\"t1.b#1\",\"t1.b#2\",\"t1.b#3\",\"t1.b#4\",\"t1.b#5\",\"t1.b#6\",\"t1.b#7\",\"t1.b#8\",\"t1.b#9\",\"t1.b#10\",\"t1.b#11\",\"t1.b#12\",\"t1.b#13\",\"t1.b#14\",\"t1.b#15\",\"t1.b#16\",\"t1.b#17\",\"t1.b#18\"]\lpk_conflict: \"NoCheck\"\l" ]
+        1 [ label = "StreamProject\lid: \"146\"\lexprs: [\"Coalesce(t1.b, 0:Int32) as $expr1\",\"Coalesce(t1.b, 0:Int32) as $expr2\",\"Coalesce(t1.b, 0:Int32) as $expr3\",\"Coalesce(t1.b, 0:Int32) as $expr4\",\"Coalesce(t1.b, 0:Int32) as $expr5\",\"Coalesce(t1.b, 0:Int32) as $expr6\",\"Coalesce(t1.b, 0:Int32) as $expr7\",\"Coalesce(t1.b, 0:Int32) as $expr8\",\"Coalesce(t1.b, 0:Int32) as $expr9\",\"Coalesce(t1.b, 0:Int32) as $expr10\",\"Coalesce(t1.b, 0:Int32) as $expr11\",\"Coalesce(t1.b, 0:Int32) as $expr12\",\"Coalesce(t1.b, 0:Int32) as $expr13\",\"Coalesce(t1.b, 0:Int32) as $expr14\",\"Coalesce(t1.b, 0:Int32) as $expr15\",\"Coalesce(t1.b, 0:Int32) as $expr16\",\"Coalesce(t1.b, 0:Int32) as $expr17\",\"Coalesce(t1.b, 0:Int32) as $expr18\",\"t1._row_id\",\"t1.b\",\"t1.a\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\",\"t1.b\"]\l" ]
+        2 [ label = "StreamHashJoin\lid: \"145\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        3 [ label = "StreamHashJoin\lid: \"137\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        4 [ label = "StreamHashJoin\lid: \"129\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        5 [ label = "StreamHashJoin\lid: \"121\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        6 [ label = "StreamHashJoin\lid: \"113\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        7 [ label = "StreamHashJoin\lid: \"105\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        8 [ label = "StreamHashJoin\lid: \"97\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        9 [ label = "StreamHashJoin\lid: \"89\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        10 [ label = "StreamHashJoin\lid: \"81\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        11 [ label = "StreamHashJoin\lid: \"73\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        12 [ label = "StreamHashJoin\lid: \"65\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        13 [ label = "StreamHashJoin\lid: \"57\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        14 [ label = "StreamHashJoin\lid: \"49\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        15 [ label = "StreamHashJoin\lid: \"41\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        16 [ label = "StreamHashJoin\lid: \"33\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        17 [ label = "StreamHashJoin\lid: \"25\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        18 [ label = "StreamHashJoin\lid: \"17\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        19 [ label = "StreamHashJoin\lid: \"9\"\ltype: \"LeftOuter\"\lpredicate: \"t1.a IS NOT DISTINCT FROM t1.a AND t1.b IS NOT DISTINCT FROM t1.b\"\l" ]
+        20 [ label = "StreamExchange\lid: \"1\"\ldist: \"HashShard(t1.a)\"\l" ]
+        21 [ label = "StreamTableScan\lid: \"12914\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        22 [ label = "StreamProject\lid: \"8\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        23 [ label = "StreamHashJoin\lid: \"7\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        24 [ label = "StreamExchange\lid: \"5\"\ldist: \"HashShard(t1.a)\"\l" ]
+        25 [ label = "StreamProject\lid: \"4\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        26 [ label = "StreamHashAgg\lid: \"3\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        27 [ label = "StreamExchange\lid: \"2\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        28 [ label = "StreamTableScan\lid: \"12921\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        29 [ label = "StreamExchange\lid: \"6\"\ldist: \"HashShard(t2.c)\"\l" ]
+        30 [ label = "StreamTableScan\lid: \"12927\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        31 [ label = "StreamProject\lid: \"16\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        32 [ label = "StreamHashJoin\lid: \"15\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        33 [ label = "StreamExchange\lid: \"13\"\ldist: \"HashShard(t1.a)\"\l" ]
+        34 [ label = "StreamProject\lid: \"12\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        35 [ label = "StreamHashAgg\lid: \"11\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        36 [ label = "StreamExchange\lid: \"10\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        37 [ label = "StreamTableScan\lid: \"12937\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        38 [ label = "StreamExchange\lid: \"14\"\ldist: \"HashShard(t2.c)\"\l" ]
+        39 [ label = "StreamTableScan\lid: \"12943\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        40 [ label = "StreamProject\lid: \"24\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        41 [ label = "StreamHashJoin\lid: \"23\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        42 [ label = "StreamExchange\lid: \"21\"\ldist: \"HashShard(t1.a)\"\l" ]
+        43 [ label = "StreamProject\lid: \"20\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        44 [ label = "StreamHashAgg\lid: \"19\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        45 [ label = "StreamExchange\lid: \"18\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        46 [ label = "StreamTableScan\lid: \"12953\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        47 [ label = "StreamExchange\lid: \"22\"\ldist: \"HashShard(t2.c)\"\l" ]
+        48 [ label = "StreamTableScan\lid: \"12959\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        49 [ label = "StreamProject\lid: \"32\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        50 [ label = "StreamHashJoin\lid: \"31\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        51 [ label = "StreamExchange\lid: \"29\"\ldist: \"HashShard(t1.a)\"\l" ]
+        52 [ label = "StreamProject\lid: \"28\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        53 [ label = "StreamHashAgg\lid: \"27\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        54 [ label = "StreamExchange\lid: \"26\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        55 [ label = "StreamTableScan\lid: \"12969\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        56 [ label = "StreamExchange\lid: \"30\"\ldist: \"HashShard(t2.c)\"\l" ]
+        57 [ label = "StreamTableScan\lid: \"12975\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        58 [ label = "StreamProject\lid: \"40\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        59 [ label = "StreamHashJoin\lid: \"39\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        60 [ label = "StreamExchange\lid: \"37\"\ldist: \"HashShard(t1.a)\"\l" ]
+        61 [ label = "StreamProject\lid: \"36\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        62 [ label = "StreamHashAgg\lid: \"35\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        63 [ label = "StreamExchange\lid: \"34\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        64 [ label = "StreamTableScan\lid: \"12985\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        65 [ label = "StreamExchange\lid: \"38\"\ldist: \"HashShard(t2.c)\"\l" ]
+        66 [ label = "StreamTableScan\lid: \"12991\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        67 [ label = "StreamProject\lid: \"48\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        68 [ label = "StreamHashJoin\lid: \"47\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        69 [ label = "StreamExchange\lid: \"45\"\ldist: \"HashShard(t1.a)\"\l" ]
+        70 [ label = "StreamProject\lid: \"44\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        71 [ label = "StreamHashAgg\lid: \"43\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        72 [ label = "StreamExchange\lid: \"42\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        73 [ label = "StreamTableScan\lid: \"13001\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        74 [ label = "StreamExchange\lid: \"46\"\ldist: \"HashShard(t2.c)\"\l" ]
+        75 [ label = "StreamTableScan\lid: \"13007\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        76 [ label = "StreamProject\lid: \"56\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        77 [ label = "StreamHashJoin\lid: \"55\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        78 [ label = "StreamExchange\lid: \"53\"\ldist: \"HashShard(t1.a)\"\l" ]
+        79 [ label = "StreamProject\lid: \"52\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        80 [ label = "StreamHashAgg\lid: \"51\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        81 [ label = "StreamExchange\lid: \"50\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        82 [ label = "StreamTableScan\lid: \"13017\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        83 [ label = "StreamExchange\lid: \"54\"\ldist: \"HashShard(t2.c)\"\l" ]
+        84 [ label = "StreamTableScan\lid: \"13023\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        85 [ label = "StreamProject\lid: \"64\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        86 [ label = "StreamHashJoin\lid: \"63\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        87 [ label = "StreamExchange\lid: \"61\"\ldist: \"HashShard(t1.a)\"\l" ]
+        88 [ label = "StreamProject\lid: \"60\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        89 [ label = "StreamHashAgg\lid: \"59\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        90 [ label = "StreamExchange\lid: \"58\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        91 [ label = "StreamTableScan\lid: \"13033\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        92 [ label = "StreamExchange\lid: \"62\"\ldist: \"HashShard(t2.c)\"\l" ]
+        93 [ label = "StreamTableScan\lid: \"13039\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        94 [ label = "StreamProject\lid: \"72\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        95 [ label = "StreamHashJoin\lid: \"71\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        96 [ label = "StreamExchange\lid: \"69\"\ldist: \"HashShard(t1.a)\"\l" ]
+        97 [ label = "StreamProject\lid: \"68\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        98 [ label = "StreamHashAgg\lid: \"67\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        99 [ label = "StreamExchange\lid: \"66\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        100 [ label = "StreamTableScan\lid: \"13049\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        101 [ label = "StreamExchange\lid: \"70\"\ldist: \"HashShard(t2.c)\"\l" ]
+        102 [ label = "StreamTableScan\lid: \"13055\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        103 [ label = "StreamProject\lid: \"80\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        104 [ label = "StreamHashJoin\lid: \"79\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        105 [ label = "StreamExchange\lid: \"77\"\ldist: \"HashShard(t1.a)\"\l" ]
+        106 [ label = "StreamProject\lid: \"76\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        107 [ label = "StreamHashAgg\lid: \"75\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        108 [ label = "StreamExchange\lid: \"74\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        109 [ label = "StreamTableScan\lid: \"13065\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        110 [ label = "StreamExchange\lid: \"78\"\ldist: \"HashShard(t2.c)\"\l" ]
+        111 [ label = "StreamTableScan\lid: \"13071\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        112 [ label = "StreamProject\lid: \"88\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        113 [ label = "StreamHashJoin\lid: \"87\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        114 [ label = "StreamExchange\lid: \"85\"\ldist: \"HashShard(t1.a)\"\l" ]
+        115 [ label = "StreamProject\lid: \"84\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        116 [ label = "StreamHashAgg\lid: \"83\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        117 [ label = "StreamExchange\lid: \"82\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        118 [ label = "StreamTableScan\lid: \"13081\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        119 [ label = "StreamExchange\lid: \"86\"\ldist: \"HashShard(t2.c)\"\l" ]
+        120 [ label = "StreamTableScan\lid: \"13087\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        121 [ label = "StreamProject\lid: \"96\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        122 [ label = "StreamHashJoin\lid: \"95\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        123 [ label = "StreamExchange\lid: \"93\"\ldist: \"HashShard(t1.a)\"\l" ]
+        124 [ label = "StreamProject\lid: \"92\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        125 [ label = "StreamHashAgg\lid: \"91\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        126 [ label = "StreamExchange\lid: \"90\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        127 [ label = "StreamTableScan\lid: \"13097\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        128 [ label = "StreamExchange\lid: \"94\"\ldist: \"HashShard(t2.c)\"\l" ]
+        129 [ label = "StreamTableScan\lid: \"13103\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        130 [ label = "StreamProject\lid: \"104\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        131 [ label = "StreamHashJoin\lid: \"103\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        132 [ label = "StreamExchange\lid: \"101\"\ldist: \"HashShard(t1.a)\"\l" ]
+        133 [ label = "StreamProject\lid: \"100\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        134 [ label = "StreamHashAgg\lid: \"99\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        135 [ label = "StreamExchange\lid: \"98\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        136 [ label = "StreamTableScan\lid: \"13113\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        137 [ label = "StreamExchange\lid: \"102\"\ldist: \"HashShard(t2.c)\"\l" ]
+        138 [ label = "StreamTableScan\lid: \"13119\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        139 [ label = "StreamProject\lid: \"112\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        140 [ label = "StreamHashJoin\lid: \"111\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        141 [ label = "StreamExchange\lid: \"109\"\ldist: \"HashShard(t1.a)\"\l" ]
+        142 [ label = "StreamProject\lid: \"108\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        143 [ label = "StreamHashAgg\lid: \"107\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        144 [ label = "StreamExchange\lid: \"106\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        145 [ label = "StreamTableScan\lid: \"13129\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        146 [ label = "StreamExchange\lid: \"110\"\ldist: \"HashShard(t2.c)\"\l" ]
+        147 [ label = "StreamTableScan\lid: \"13135\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        148 [ label = "StreamProject\lid: \"120\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        149 [ label = "StreamHashJoin\lid: \"119\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        150 [ label = "StreamExchange\lid: \"117\"\ldist: \"HashShard(t1.a)\"\l" ]
+        151 [ label = "StreamProject\lid: \"116\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        152 [ label = "StreamHashAgg\lid: \"115\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        153 [ label = "StreamExchange\lid: \"114\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        154 [ label = "StreamTableScan\lid: \"13145\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        155 [ label = "StreamExchange\lid: \"118\"\ldist: \"HashShard(t2.c)\"\l" ]
+        156 [ label = "StreamTableScan\lid: \"13151\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        157 [ label = "StreamProject\lid: \"128\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        158 [ label = "StreamHashJoin\lid: \"127\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        159 [ label = "StreamExchange\lid: \"125\"\ldist: \"HashShard(t1.a)\"\l" ]
+        160 [ label = "StreamProject\lid: \"124\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        161 [ label = "StreamHashAgg\lid: \"123\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        162 [ label = "StreamExchange\lid: \"122\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        163 [ label = "StreamTableScan\lid: \"13161\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        164 [ label = "StreamExchange\lid: \"126\"\ldist: \"HashShard(t2.c)\"\l" ]
+        165 [ label = "StreamTableScan\lid: \"13167\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        166 [ label = "StreamProject\lid: \"136\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        167 [ label = "StreamHashJoin\lid: \"135\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        168 [ label = "StreamExchange\lid: \"133\"\ldist: \"HashShard(t1.a)\"\l" ]
+        169 [ label = "StreamProject\lid: \"132\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        170 [ label = "StreamHashAgg\lid: \"131\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        171 [ label = "StreamExchange\lid: \"130\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        172 [ label = "StreamTableScan\lid: \"13177\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        173 [ label = "StreamExchange\lid: \"134\"\ldist: \"HashShard(t2.c)\"\l" ]
+        174 [ label = "StreamTableScan\lid: \"13183\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        175 [ label = "StreamProject\lid: \"144\"\lexprs: [\"t1.a\",\"t1.b\",\"t1.b\"]\l" ]
+        176 [ label = "StreamHashJoin\lid: \"143\"\ltype: \"Inner\"\lpredicate: \"t1.a = t2.c\"\l" ]
+        177 [ label = "StreamExchange\lid: \"141\"\ldist: \"HashShard(t1.a)\"\l" ]
+        178 [ label = "StreamProject\lid: \"140\"\lexprs: [\"t1.a\",\"t1.b\"]\l" ]
+        179 [ label = "StreamHashAgg\lid: \"139\"\lgroup_key: [\"t1.a\",\"t1.b\"]\laggs: [\"count\"]\l" ]
+        180 [ label = "StreamExchange\lid: \"138\"\ldist: \"HashShard(t1.a, t1.b)\"\l" ]
+        181 [ label = "StreamTableScan\lid: \"13193\"\ltable: \"t1\"\lcolumns: [\"a\",\"b\",\"_row_id\"]\l" ]
+        182 [ label = "StreamExchange\lid: \"142\"\ldist: \"HashShard(t2.c)\"\l" ]
+        183 [ label = "StreamTableScan\lid: \"13199\"\ltable: \"t2\"\lcolumns: [\"c\"]\l" ]
+        0 -> 1 [ ]
+        1 -> 2 [ ]
+        2 -> 3 [ ]
+        3 -> 4 [ ]
+        4 -> 5 [ ]
+        5 -> 6 [ ]
+        6 -> 7 [ ]
+        7 -> 8 [ ]
+        8 -> 9 [ ]
+        9 -> 10 [ ]
+        10 -> 11 [ ]
+        11 -> 12 [ ]
+        12 -> 13 [ ]
+        13 -> 14 [ ]
+        14 -> 15 [ ]
+        15 -> 16 [ ]
+        16 -> 17 [ ]
+        17 -> 18 [ ]
+        18 -> 19 [ ]
+        19 -> 20 [ ]
+        20 -> 21 [ ]
+        19 -> 22 [ ]
+        22 -> 23 [ ]
+        23 -> 24 [ ]
+        24 -> 25 [ ]
+        25 -> 26 [ ]
+        26 -> 27 [ ]
+        27 -> 28 [ ]
+        23 -> 29 [ ]
+        29 -> 30 [ ]
+        18 -> 31 [ ]
+        31 -> 32 [ ]
+        32 -> 33 [ ]
+        33 -> 34 [ ]
+        34 -> 35 [ ]
+        35 -> 36 [ ]
+        36 -> 37 [ ]
+        32 -> 38 [ ]
+        38 -> 39 [ ]
+        17 -> 40 [ ]
+        40 -> 41 [ ]
+        41 -> 42 [ ]
+        42 -> 43 [ ]
+        43 -> 44 [ ]
+        44 -> 45 [ ]
+        45 -> 46 [ ]
+        41 -> 47 [ ]
+        47 -> 48 [ ]
+        16 -> 49 [ ]
+        49 -> 50 [ ]
+        50 -> 51 [ ]
+        51 -> 52 [ ]
+        52 -> 53 [ ]
+        53 -> 54 [ ]
+        54 -> 55 [ ]
+        50 -> 56 [ ]
+        56 -> 57 [ ]
+        15 -> 58 [ ]
+        58 -> 59 [ ]
+        59 -> 60 [ ]
+        60 -> 61 [ ]
+        61 -> 62 [ ]
+        62 -> 63 [ ]
+        63 -> 64 [ ]
+        59 -> 65 [ ]
+        65 -> 66 [ ]
+        14 -> 67 [ ]
+        67 -> 68 [ ]
+        68 -> 69 [ ]
+        69 -> 70 [ ]
+        70 -> 71 [ ]
+        71 -> 72 [ ]
+        72 -> 73 [ ]
+        68 -> 74 [ ]
+        74 -> 75 [ ]
+        13 -> 76 [ ]
+        76 -> 77 [ ]
+        77 -> 78 [ ]
+        78 -> 79 [ ]
+        79 -> 80 [ ]
+        80 -> 81 [ ]
+        81 -> 82 [ ]
+        77 -> 83 [ ]
+        83 -> 84 [ ]
+        12 -> 85 [ ]
+        85 -> 86 [ ]
+        86 -> 87 [ ]
+        87 -> 88 [ ]
+        88 -> 89 [ ]
+        89 -> 90 [ ]
+        90 -> 91 [ ]
+        86 -> 92 [ ]
+        92 -> 93 [ ]
+        11 -> 94 [ ]
+        94 -> 95 [ ]
+        95 -> 96 [ ]
+        96 -> 97 [ ]
+        97 -> 98 [ ]
+        98 -> 99 [ ]
+        99 -> 100 [ ]
+        95 -> 101 [ ]
+        101 -> 102 [ ]
+        10 -> 103 [ ]
+        103 -> 104 [ ]
+        104 -> 105 [ ]
+        105 -> 106 [ ]
+        106 -> 107 [ ]
+        107 -> 108 [ ]
+        108 -> 109 [ ]
+        104 -> 110 [ ]
+        110 -> 111 [ ]
+        9 -> 112 [ ]
+        112 -> 113 [ ]
+        113 -> 114 [ ]
+        114 -> 115 [ ]
+        115 -> 116 [ ]
+        116 -> 117 [ ]
+        117 -> 118 [ ]
+        113 -> 119 [ ]
+        119 -> 120 [ ]
+        8 -> 121 [ ]
+        121 -> 122 [ ]
+        122 -> 123 [ ]
+        123 -> 124 [ ]
+        124 -> 125 [ ]
+        125 -> 126 [ ]
+        126 -> 127 [ ]
+        122 -> 128 [ ]
+        128 -> 129 [ ]
+        7 -> 130 [ ]
+        130 -> 131 [ ]
+        131 -> 132 [ ]
+        132 -> 133 [ ]
+        133 -> 134 [ ]
+        134 -> 135 [ ]
+        135 -> 136 [ ]
+        131 -> 137 [ ]
+        137 -> 138 [ ]
+        6 -> 139 [ ]
+        139 -> 140 [ ]
+        140 -> 141 [ ]
+        141 -> 142 [ ]
+        142 -> 143 [ ]
+        143 -> 144 [ ]
+        144 -> 145 [ ]
+        140 -> 146 [ ]
+        146 -> 147 [ ]
+        5 -> 148 [ ]
+        148 -> 149 [ ]
+        149 -> 150 [ ]
+        150 -> 151 [ ]
+        151 -> 152 [ ]
+        152 -> 153 [ ]
+        153 -> 154 [ ]
+        149 -> 155 [ ]
+        155 -> 156 [ ]
+        4 -> 157 [ ]
+        157 -> 158 [ ]
+        158 -> 159 [ ]
+        159 -> 160 [ ]
+        160 -> 161 [ ]
+        161 -> 162 [ ]
+        162 -> 163 [ ]
+        158 -> 164 [ ]
+        164 -> 165 [ ]
+        3 -> 166 [ ]
+        166 -> 167 [ ]
+        167 -> 168 [ ]
+        168 -> 169 [ ]
+        169 -> 170 [ ]
+        170 -> 171 [ ]
+        171 -> 172 [ ]
+        167 -> 173 [ ]
+        173 -> 174 [ ]
+        2 -> 175 [ ]
+        175 -> 176 [ ]
+        176 -> 177 [ ]
+        177 -> 178 [ ]
+        178 -> 179 [ ]
+        179 -> 180 [ ]
+        180 -> 181 [ ]
+        176 -> 182 [ ]
+        182 -> 183 [ ]
+    }