Skip to content

Commit 6940ae4

Browse files
ice1000ice1000
authored
feat(pretty): introduce the proposed pretty printer in stream_graph_formatter (risingwavelabs#8576)
1 parent 99f4604 commit 6940ae4

14 files changed

+5564
-3247
lines changed

‎Cargo.lock

Lines changed: 7 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎src/frontend/Cargo.toml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ petgraph = "0.6"
4444
pgwire = { path = "../utils/pgwire" }
4545
pin-project-lite = "0.2"
4646
postgres-types = { version = "0.2.4" }
47+
pretty-xmlish = "0.1.10"
4748
prometheus = { version = "0.13", features = ["process"] }
4849
rand = "0.8"
4950
risingwave_batch = { path = "../batch" }

‎src/frontend/planner_test/src/lib.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -635,7 +635,7 @@ impl TestCase {
635635
// Only generate stream_dist_plan if it is specified in test case
636636
if self.stream_dist_plan.is_some() {
637637
let graph = build_graph(stream_plan);
638-
ret.stream_dist_plan = Some(explain_stream_graph(&graph, false).unwrap());
638+
ret.stream_dist_plan = Some(explain_stream_graph(&graph, false));
639639
}
640640
}
641641
}

‎src/frontend/planner_test/tests/testdata/distribution_derive.yaml

Lines changed: 666 additions & 309 deletions
Large diffs are not rendered by default.

‎src/frontend/planner_test/tests/testdata/join.yaml

Lines changed: 55 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -638,66 +638,68 @@
638638
| └─StreamTableScan { table: t, columns: [t.src, t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
639639
└─StreamExchange { dist: HashShard(t.dst) }
640640
└─StreamTableScan { table: t, columns: [t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
641-
stream_dist_plan: |
641+
stream_dist_plan: |+
642642
Fragment 0
643-
StreamMaterialize { columns: [p1, p2, p3, t._row_id(hidden), t._row_id#1(hidden), t.src(hidden), t._row_id#2(hidden)], pk_columns: [t._row_id, t._row_id#1, p2, t._row_id#2, t.src, p1], pk_conflict: "no check" }
644-
materialized table: 4294967294
645-
StreamHashJoin { type: Inner, predicate: t.src = t.dst AND t.src = t.dst, output: [t.src, t.dst, t.dst, t._row_id, t._row_id, t.src, t._row_id] }
646-
left table: 0, right table 2, left degree table: 1, right degree table: 3,
647-
StreamExchange Hash([0]) from 1
648-
StreamExchange Hash([0]) from 4
643+
StreamMaterialize { columns: [p1, p2, p3, t._row_id(hidden), t._row_id#1(hidden), t.src(hidden), t._row_id#2(hidden)], pk_columns: [t._row_id, t._row_id#1, p2, t._row_id#2, t.src, p1], pk_conflict: "no check" }
644+
├── materialized table: 4294967294
645+
└── StreamHashJoin { type: Inner, predicate: t.src = t.dst AND t.src = t.dst, output: [t.src, t.dst, t.dst, t._row_id, t._row_id, t.src, t._row_id] }
646+
├── left table: 0
647+
├── right table: 2
648+
├── left degree table: 1
649+
├── right degree table: 3
650+
├── StreamExchange Hash([0]) from 1
651+
└── StreamExchange Hash([0]) from 4
649652
650653
Fragment 1
651-
StreamHashJoin { type: Inner, predicate: t.dst = t.src, output: [t.src, t.dst, t.src, t.dst, t._row_id, t._row_id] }
652-
left table: 4, right table 6, left degree table: 5, right degree table: 7,
653-
StreamExchange Hash([1]) from 2
654-
StreamExchange Hash([0]) from 3
654+
StreamHashJoin { type: Inner, predicate: t.dst = t.src, output: [t.src, t.dst, t.src, t.dst, t._row_id, t._row_id] } { left table: 4, right table: 6, left degree table: 5, right degree table: 7 }
655+
├── StreamExchange Hash([1]) from 2
656+
└── StreamExchange Hash([0]) from 3
655657
656658
Fragment 2
657-
Chain { table: t, columns: [t.src, t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
658-
Upstream
659-
BatchPlanNode
659+
Chain { table: t, columns: [t.src, t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
660+
├── Upstream
661+
└── BatchPlanNode
660662
661663
Fragment 3
662-
Chain { table: t, columns: [t.src, t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
663-
Upstream
664-
BatchPlanNode
664+
Chain { table: t, columns: [t.src, t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
665+
├── Upstream
666+
└── BatchPlanNode
665667
666668
Fragment 4
667-
Chain { table: t, columns: [t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
668-
Upstream
669-
BatchPlanNode
669+
Chain { table: t, columns: [t.dst, t._row_id], pk: [t._row_id], dist: UpstreamHashShard(t._row_id) }
670+
├── Upstream
671+
└── BatchPlanNode
672+
673+
Table 0
674+
├── columns: [ t_src, t_dst, t_src_0, t_dst_0, t__row_id, t__row_id_0 ]
675+
├── primary key: [ $2 ASC, $0 ASC, $4 ASC, $5 ASC, $1 ASC ]
676+
├── value indices: [ 0, 1, 2, 3, 4, 5 ]
677+
├── distribution key: [ 0 ]
678+
└── read pk prefix len hint: 2
679+
680+
Table 1
681+
├── columns: [ t_src, t_src_0, t__row_id, t__row_id_0, t_dst, _degree ]
682+
├── primary key: [ $0 ASC, $1 ASC, $2 ASC, $3 ASC, $4 ASC ]
683+
├── value indices: [ 5 ]
684+
├── distribution key: [ 1 ]
685+
└── read pk prefix len hint: 2
686+
687+
Table 2 { columns: [ t_dst, t__row_id ], primary key: [ $0 ASC, $0 ASC, $1 ASC ], value indices: [ 0, 1 ], distribution key: [ 0 ], read pk prefix len hint: 2 }
688+
689+
Table 3 { columns: [ t_dst, t_dst_0, t__row_id, _degree ], primary key: [ $0 ASC, $1 ASC, $2 ASC ], value indices: [ 3 ], distribution key: [ 1 ], read pk prefix len hint: 2 }
690+
691+
Table 4 { columns: [ t_src, t_dst, t__row_id ], primary key: [ $1 ASC, $2 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 1 ], read pk prefix len hint: 1 }
692+
693+
Table 5 { columns: [ t_dst, t__row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }
694+
695+
Table 6 { columns: [ t_src, t_dst, t__row_id ], primary key: [ $0 ASC, $2 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }
696+
697+
Table 7 { columns: [ t_src, t__row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }
698+
699+
Table 4294967294
700+
├── columns: [ p1, p2, p3, t._row_id, t._row_id#1, t.src, t._row_id#2 ]
701+
├── primary key: [ $3 ASC, $4 ASC, $1 ASC, $6 ASC, $5 ASC, $0 ASC ]
702+
├── value indices: [ 0, 1, 2, 3, 4, 5, 6 ]
703+
├── distribution key: [ 0 ]
704+
└── read pk prefix len hint: 6
670705
671-
Table 0 { columns: [t_src, t_dst, t_src_0, t_dst_0, t__row_id, t__row_id_0], primary key: [$2 ASC, $0 ASC, $4 ASC, $5 ASC, $1 ASC], value indices: [0, 1, 2, 3, 4, 5], distribution key: [0], read pk prefix len hint: 2 }
672-
Table 1 { columns: [t_src, t_src_0, t__row_id, t__row_id_0, t_dst, _degree], primary key: [$0 ASC, $1 ASC, $2 ASC, $3 ASC, $4 ASC], value indices: [5], distribution key: [1], read pk prefix len hint: 2 }
673-
Table 2 { columns: [t_dst, t__row_id], primary key: [$0 ASC, $0 ASC, $1 ASC], value indices: [0, 1], distribution key: [0], read pk prefix len hint: 2 }
674-
Table 3 { columns: [t_dst, t_dst_0, t__row_id, _degree], primary key: [$0 ASC, $1 ASC, $2 ASC], value indices: [3], distribution key: [1], read pk prefix len hint: 2 }
675-
Table 4 { columns: [t_src, t_dst, t__row_id], primary key: [$1 ASC, $2 ASC], value indices: [0, 1, 2], distribution key: [1], read pk prefix len hint: 1 }
676-
Table 5 { columns: [t_dst, t__row_id, _degree], primary key: [$0 ASC, $1 ASC], value indices: [2], distribution key: [0], read pk prefix len hint: 1 }
677-
Table 6 { columns: [t_src, t_dst, t__row_id], primary key: [$0 ASC, $2 ASC], value indices: [0, 1, 2], distribution key: [0], read pk prefix len hint: 1 }
678-
Table 7 { columns: [t_src, t__row_id, _degree], primary key: [$0 ASC, $1 ASC], value indices: [2], distribution key: [0], read pk prefix len hint: 1 }
679-
Table 4294967294 { columns: [p1, p2, p3, t._row_id, t._row_id#1, t.src, t._row_id#2], primary key: [$3 ASC, $4 ASC, $1 ASC, $6 ASC, $5 ASC, $0 ASC], value indices: [0, 1, 2, 3, 4, 5, 6], distribution key: [0], read pk prefix len hint: 6 }
680-
- name: Fix hash join distribution key (https://github.com/risingwavelabs/risingwave/issues/8537)
681-
sql: |
682-
CREATE TABLE part (
683-
p INTEGER,
684-
c VARCHAR,
685-
PRIMARY KEY (p)
686-
);
687-
CREATE TABLE B (
688-
b INTEGER,
689-
d VARCHAR,
690-
PRIMARY KEY (b)
691-
);
692-
select B.* from part join B on part.c = B.d join part p1 on p1.p = part.p and p1.p = B.b;
693-
stream_plan: |
694-
StreamMaterialize { columns: [b, d, part.p(hidden), part.c(hidden), part.p#1(hidden)], pk_columns: [part.p, b, part.c, part.p#1], pk_conflict: "no check" }
695-
└─StreamHashJoin { type: Inner, predicate: part.p = part.p AND b.b = part.p, output: [b.b, b.d, part.p, part.c, part.p] }
696-
├─StreamExchange { dist: HashShard(part.p, b.b) }
697-
| └─StreamHashJoin { type: Inner, predicate: part.c = b.d, output: [part.p, b.b, b.d, part.c] }
698-
| ├─StreamExchange { dist: HashShard(part.c) }
699-
| | └─StreamTableScan { table: part, columns: [part.p, part.c], pk: [part.p], dist: UpstreamHashShard(part.p) }
700-
| └─StreamExchange { dist: HashShard(b.d) }
701-
| └─StreamTableScan { table: b, columns: [b.b, b.d], pk: [b.b], dist: UpstreamHashShard(b.b) }
702-
└─StreamExchange { dist: HashShard(part.p, part.p) }
703-
└─StreamTableScan { table: part, columns: [part.p], pk: [part.p], dist: UpstreamHashShard(part.p) }

0 commit comments

Comments
 (0)