Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor: cherry pick #12524 and #12840 to v1.3-rc #12888

Merged
merged 2 commits into from
Oct 17, 2023
Merged

refactor: cherry pick #12524 and #12840 to v1.3-rc #12888

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e2936bc
refactor(optimizer): change stream key to option (#12524)
st1page Oct 10, 2023
0ab1456
fix(stream topn): fix pk indices of `TopN` and `GroupTopN` (#12840)
stdrc Oct 16, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions src/frontend/planner_test/tests/testdata/output/nexmark.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1600,13 +1600,13 @@
└─BatchExchange { order: [], dist: HashShard(bid.bidder, bid.auction) }
└─BatchScan { table: bid, columns: [bid.auction, bid.bidder, bid.price, bid.channel, bid.url, bid.date_time, bid.extra], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, bid._row_id(hidden)], stream_key: [bid._row_id, bidder, auction], pk_columns: [bid._row_id, bidder, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, bid._row_id(hidden)], stream_key: [bidder, auction, bid._row_id], pk_columns: [bidder, auction, bid._row_id], pk_conflict: NoCheck }
└─StreamGroupTopN [append_only] { order: [bid.date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [bid.bidder, bid.auction] }
└─StreamExchange { dist: HashShard(bid.bidder, bid.auction) }
└─StreamTableScan { table: bid, columns: [bid.auction, bid.bidder, bid.price, bid.channel, bid.url, bid.date_time, bid.extra, bid._row_id], pk: [bid._row_id], dist: UpstreamHashShard(bid._row_id) }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, bid._row_id(hidden)], stream_key: [bid._row_id, bidder, auction], pk_columns: [bid._row_id, bidder, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, bid._row_id(hidden)], stream_key: [bidder, auction, bid._row_id], pk_columns: [bidder, auction, bid._row_id], pk_conflict: NoCheck }
├── materialized table: 4294967294
└── StreamGroupTopN [append_only] { order: [bid.date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [bid.bidder, bid.auction] } { state table: 0 }
└── StreamExchange Hash([1, 0]) from 1
Expand All @@ -1627,7 +1627,7 @@

Table 4294967294
├── columns: [ auction, bidder, price, channel, url, date_time, extra, bid._row_id ]
├── primary key: [ $7 ASC, $1 ASC, $0 ASC ]
├── primary key: [ $1 ASC, $0 ASC, $7 ASC ]
├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7 ]
├── distribution key: [ 1, 0 ]
└── read pk prefix len hint: 3
Expand All @@ -1652,13 +1652,13 @@
└─BatchExchange { order: [], dist: HashShard(bid.auction) }
└─BatchScan { table: bid, columns: [bid.auction, bid.bidder, bid.price, bid.channel, bid.url, bid.date_time, bid.extra, bid.p_time], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden)], stream_key: [bid._row_id, auction], pk_columns: [bid._row_id, auction], pk_conflict: NoCheck, watermark_columns: [p_time] }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden)], stream_key: [auction, bid._row_id], pk_columns: [auction, bid._row_id], pk_conflict: NoCheck, watermark_columns: [p_time] }
└─StreamGroupTopN [append_only] { order: [bid.price DESC], limit: 10, offset: 0, group_key: [bid.auction], output_watermarks: [bid.p_time] }
└─StreamExchange { dist: HashShard(bid.auction) }
└─StreamTableScan { table: bid, columns: [bid.auction, bid.bidder, bid.price, bid.channel, bid.url, bid.date_time, bid.extra, bid.p_time, bid._row_id], pk: [bid._row_id], dist: UpstreamHashShard(bid._row_id) }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden)], stream_key: [bid._row_id, auction], pk_columns: [bid._row_id, auction], pk_conflict: NoCheck, watermark_columns: [p_time] }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden)], stream_key: [auction, bid._row_id], pk_columns: [auction, bid._row_id], pk_conflict: NoCheck, watermark_columns: [p_time] }
├── materialized table: 4294967294
└── StreamGroupTopN [append_only] { order: [bid.price DESC], limit: 10, offset: 0, group_key: [bid.auction], output_watermarks: [bid.p_time] } { state table: 0 }
└── StreamExchange Hash([0]) from 1
Expand All @@ -1679,7 +1679,7 @@

Table 4294967294
├── columns: [ auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id ]
├── primary key: [ $8 ASC, $0 ASC ]
├── primary key: [ $0 ASC, $8 ASC ]
├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8 ]
├── distribution key: [ 0 ]
└── read pk prefix len hint: 2
Expand All @@ -1706,14 +1706,14 @@
└─BatchExchange { order: [], dist: HashShard(bid.auction) }
└─BatchScan { table: bid, columns: [bid.auction, bid.bidder, bid.price, bid.channel, bid.url, bid.date_time, bid.extra, bid.p_time], distribution: SomeShard }
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden), rank_number], stream_key: [bid._row_id, auction], pk_columns: [bid._row_id, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden), rank_number], stream_key: [auction, bid._row_id], pk_columns: [auction, bid._row_id], pk_conflict: NoCheck }
└─StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY bid.auction ORDER BY bid.price DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] }
└─StreamGroupTopN [append_only] { order: [bid.price DESC], limit: 10, offset: 0, group_key: [bid.auction], output_watermarks: [bid.p_time] }
└─StreamExchange { dist: HashShard(bid.auction) }
└─StreamTableScan { table: bid, columns: [bid.auction, bid.bidder, bid.price, bid.channel, bid.url, bid.date_time, bid.extra, bid.p_time, bid._row_id], pk: [bid._row_id], dist: UpstreamHashShard(bid._row_id) }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden), rank_number], stream_key: [bid._row_id, auction], pk_columns: [bid._row_id, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id(hidden), rank_number], stream_key: [auction, bid._row_id], pk_columns: [auction, bid._row_id], pk_conflict: NoCheck }
├── materialized table: 4294967294
└── StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY bid.auction ORDER BY bid.price DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] } { state table: 0 }
└── StreamGroupTopN [append_only] { order: [bid.price DESC], limit: 10, offset: 0, group_key: [bid.auction], output_watermarks: [bid.p_time] } { state table: 1 }
Expand Down Expand Up @@ -1742,7 +1742,7 @@

Table 4294967294
├── columns: [ auction, bidder, price, channel, url, date_time, extra, p_time, bid._row_id, rank_number ]
├── primary key: [ $8 ASC, $0 ASC ]
├── primary key: [ $0 ASC, $8 ASC ]
├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
├── distribution key: [ 0 ]
└── read pk prefix len hint: 2
Expand Down Expand Up @@ -2447,7 +2447,7 @@

Table 0
├── columns: [ auction_id, first_value(auction_item_name order_by(auction_item_name ASC)), count(bid_auction), $expr1 ]
├── primary key: [ $2 DESC, $0 ASC, $3 ASC ]
├── primary key: [ $2 DESC, $0 ASC ]
├── value indices: [ 0, 1, 2, 3 ]
├── distribution key: []
└── read pk prefix len hint: 0
Expand Down
14 changes: 7 additions & 7 deletions src/frontend/planner_test/tests/testdata/output/nexmark_source.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1499,15 +1499,15 @@
└─BatchProject { exprs: [auction, bidder, price, channel, url, date_time, extra] }
└─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [_row_id, bidder, auction], pk_columns: [_row_id, bidder, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [bidder, auction, _row_id], pk_columns: [bidder, auction, _row_id], pk_conflict: NoCheck }
└─StreamGroupTopN [append_only] { order: [date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [bidder, auction] }
└─StreamExchange { dist: HashShard(bidder, auction) }
└─StreamProject { exprs: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
└─StreamRowIdGen { row_id_index: 7 }
└─StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [_row_id, bidder, auction], pk_columns: [_row_id, bidder, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [bidder, auction, _row_id], pk_columns: [bidder, auction, _row_id], pk_conflict: NoCheck }
├── materialized table: 4294967294
└── StreamGroupTopN [append_only] { order: [date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [bidder, auction] } { state table: 0 }
└── StreamExchange Hash([1, 0]) from 1
Expand All @@ -1528,7 +1528,7 @@

Table 4294967294
├── columns: [ auction, bidder, price, channel, url, date_time, extra, _row_id ]
├── primary key: [ $7 ASC, $1 ASC, $0 ASC ]
├── primary key: [ $1 ASC, $0 ASC, $7 ASC ]
├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7 ]
├── distribution key: [ 1, 0 ]
└── read pk prefix len hint: 3
Expand Down Expand Up @@ -1556,7 +1556,7 @@
└─BatchProject { exprs: [auction, bidder, price, channel, url, date_time, extra] }
└─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden), rank_number], stream_key: [_row_id, auction], pk_columns: [_row_id, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden), rank_number], stream_key: [auction, _row_id], pk_columns: [auction, _row_id], pk_conflict: NoCheck }
└─StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY auction ORDER BY price DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] }
└─StreamGroupTopN [append_only] { order: [price DESC], limit: 10, offset: 0, group_key: [auction] }
└─StreamExchange { dist: HashShard(auction) }
Expand All @@ -1565,7 +1565,7 @@
└─StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden), rank_number], stream_key: [_row_id, auction], pk_columns: [_row_id, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden), rank_number], stream_key: [auction, _row_id], pk_columns: [auction, _row_id], pk_conflict: NoCheck }
├── materialized table: 4294967294
└── StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY auction ORDER BY price DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] } { state table: 0 }
└── StreamGroupTopN [append_only] { order: [price DESC], limit: 10, offset: 0, group_key: [auction] } { state table: 1 }
Expand Down Expand Up @@ -1594,7 +1594,7 @@

Table 4294967294
├── columns: [ auction, bidder, price, channel, url, date_time, extra, _row_id, rank_number ]
├── primary key: [ $7 ASC, $0 ASC ]
├── primary key: [ $0 ASC, $7 ASC ]
├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8 ]
├── distribution key: [ 0 ]
└── read pk prefix len hint: 2
Expand Down Expand Up @@ -2199,7 +2199,7 @@
StreamRowIdGen { row_id_index: 7 }
└── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 8 }

Table 0 { columns: [ id, item_name, count(auction), $expr1 ], primary key: [ $2 DESC, $0 ASC, $1 ASC, $3 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [], read pk prefix len hint: 0 }
Table 0 { columns: [ id, item_name, count(auction), $expr1 ], primary key: [ $2 DESC, $0 ASC, $1 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [], read pk prefix len hint: 0 }

Table 1
├── columns: [ id, item_name, count(auction), $expr1 ]
Expand Down
14 changes: 7 additions & 7 deletions src/frontend/planner_test/tests/testdata/output/nexmark_temporal_filter.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1051,7 +1051,7 @@
FROM bid)
WHERE rank_number <= 1;
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [_row_id, bidder, auction], pk_columns: [_row_id, bidder, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [bidder, auction, _row_id], pk_columns: [bidder, auction, _row_id], pk_conflict: NoCheck }
└─StreamGroupTopN { order: [$expr8 DESC], limit: 1, offset: 0, with_ties: true, group_key: [$expr4, $expr3] }
└─StreamExchange { dist: HashShard($expr4, $expr3) }
└─StreamProject { exprs: [Field(bid, 0:Int32) as $expr3, Field(bid, 1:Int32) as $expr4, Field(bid, 2:Int32) as $expr5, Field(bid, 3:Int32) as $expr6, Field(bid, 4:Int32) as $expr7, Field(bid, 5:Int32) as $expr8, Field(bid, 6:Int32) as $expr9, _row_id] }
Expand All @@ -1065,7 +1065,7 @@
└─StreamNow { output: [now] }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [_row_id, bidder, auction], pk_columns: [_row_id, bidder, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, _row_id(hidden)], stream_key: [bidder, auction, _row_id], pk_columns: [bidder, auction, _row_id], pk_conflict: NoCheck }
├── materialized table: 4294967294
└── StreamGroupTopN { order: [$expr8 DESC], limit: 1, offset: 0, with_ties: true, group_key: [$expr4, $expr3] } { state table: 0 }
└── StreamExchange Hash([1, 0]) from 1
Expand Down Expand Up @@ -1093,7 +1093,7 @@

Table 4 { columns: [ now ], primary key: [], value indices: [ 0 ], distribution key: [], read pk prefix len hint: 0 }

Table 4294967294 { columns: [ auction, bidder, price, channel, url, date_time, extra, _row_id ], primary key: [ $7 ASC, $1 ASC, $0 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7 ], distribution key: [ 1, 0 ], read pk prefix len hint: 3 }
Table 4294967294 { columns: [ auction, bidder, price, channel, url, date_time, extra, _row_id ], primary key: [ $1 ASC, $0 ASC, $7 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7 ], distribution key: [ 1, 0 ], read pk prefix len hint: 3 }

- id: nexmark_q19
before:
Expand All @@ -1103,7 +1103,7 @@
(SELECT *, ROW_NUMBER() OVER (PARTITION BY auction ORDER BY price DESC) AS rank_number FROM bid)
WHERE rank_number <= 10;
stream_plan: |-
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, _row_id(hidden), rank_number], stream_key: [_row_id, auction], pk_columns: [_row_id, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, _row_id(hidden), rank_number], stream_key: [auction, _row_id], pk_columns: [auction, _row_id], pk_conflict: NoCheck }
└─StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY $expr3 ORDER BY $expr5 DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] }
└─StreamGroupTopN { order: [$expr5 DESC], limit: 10, offset: 0, group_key: [$expr3] }
└─StreamExchange { dist: HashShard($expr3) }
Expand All @@ -1118,7 +1118,7 @@
└─StreamNow { output: [now] }
stream_dist_plan: |+
Fragment 0
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, _row_id(hidden), rank_number], stream_key: [_row_id, auction], pk_columns: [_row_id, auction], pk_conflict: NoCheck }
StreamMaterialize { columns: [auction, bidder, price, channel, url, date_time, extra, p_time, _row_id(hidden), rank_number], stream_key: [auction, _row_id], pk_columns: [auction, _row_id], pk_conflict: NoCheck }
├── materialized table: 4294967294
└── StreamOverWindow { window_functions: [row_number() OVER(PARTITION BY $expr3 ORDER BY $expr5 DESC ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)] } { state table: 0 }
└── StreamGroupTopN { order: [$expr5 DESC], limit: 10, offset: 0, group_key: [$expr3] } { state table: 1 }
Expand Down Expand Up @@ -1149,7 +1149,7 @@

Table 5 { columns: [ now ], primary key: [], value indices: [ 0 ], distribution key: [], read pk prefix len hint: 0 }

Table 4294967294 { columns: [ auction, bidder, price, channel, url, date_time, extra, p_time, _row_id, rank_number ], primary key: [ $8 ASC, $0 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ], distribution key: [ 0 ], read pk prefix len hint: 2 }
Table 4294967294 { columns: [ auction, bidder, price, channel, url, date_time, extra, p_time, _row_id, rank_number ], primary key: [ $0 ASC, $8 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ], distribution key: [ 0 ], read pk prefix len hint: 2 }

- id: nexmark_q20
before:
Expand Down Expand Up @@ -1919,7 +1919,7 @@
StreamProject { exprs: [SubtractWithTimeZone(now, '00:05:00':Interval, 'UTC':Varchar) as $expr4], output_watermarks: [$expr4] }
└── StreamNow { output: [now] } { state table: 10 }

Table 0 { columns: [ $expr2, $expr3, count($expr5), $expr6 ], primary key: [ $2 DESC, $0 ASC, $1 ASC, $3 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [], read pk prefix len hint: 0 }
Table 0 { columns: [ $expr2, $expr3, count($expr5), $expr6 ], primary key: [ $2 DESC, $0 ASC, $1 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [], read pk prefix len hint: 0 }

Table 1
├── columns: [ $expr2, $expr3, count($expr5), $expr6 ]
Expand Down
Loading
Loading