nexmark_source.yaml

# This file is automatically generated. See `src/frontend/planner_test/README.md` for more information.
- id: create_sources
  sql: |
    create source auction (
      id BIGINT,
      item_name VARCHAR,
      description VARCHAR,
      initial_bid BIGINT,
      reserve BIGINT,
      date_time TIMESTAMP,
      expires TIMESTAMP,
      seller BIGINT,
      category BIGINT,
      extra VARCHAR)
    with (
      connector = 'nexmark',
      nexmark.table.type = 'Auction'
    );

    create source bid (
      auction BIGINT,
      bidder BIGINT,
      price BIGINT,
      channel VARCHAR,
      url VARCHAR,
      date_time TIMESTAMP,
      extra VARCHAR)
    with (
      connector = 'nexmark',
      nexmark.table.type = 'Bid'
    );

    create source person (
      id BIGINT,
      name VARCHAR,
      email_address VARCHAR,
      credit_card VARCHAR,
      city VARCHAR,
      state VARCHAR,
      date_time TIMESTAMP,
      extra VARCHAR)
    with (
      connector = 'nexmark',
      nexmark.table.type = 'Person'
    );
- id: nexmark_q0
  before:
  - create_sources
  sql: |
    SELECT auction, bidder, price, date_time FROM bid;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, bidder, price, date_time] }
      └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, bidder, price, date_time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, bidder, price, date_time, _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, date_time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [auction, bidder, price, date_time, _row_id] }
        └── StreamRowIdGen { row_id_index: 7 }
            └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction, bidder, price, date_time, _row_id ]
    ├── primary key: [ $4 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4 ]
    ├── distribution key: [ 4 ]
    └── read pk prefix len hint: 1

- id: nexmark_q1
  before:
  - create_sources
  sql: |
    SELECT
      auction,
      bidder,
      0.908 * price as price,
      date_time
    FROM bid;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, bidder, (0.908:Decimal * price::Decimal) as $expr1, date_time] }
      └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, bidder, price, date_time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, bidder, (0.908:Decimal * price::Decimal) as $expr1, date_time, _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, date_time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [auction, bidder, (0.908:Decimal * price::Decimal) as $expr1, date_time, _row_id] }
        └── StreamRowIdGen { row_id_index: 7 }
            └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction, bidder, price, date_time, _row_id ]
    ├── primary key: [ $4 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4 ]
    ├── distribution key: [ 4 ]
    └── read pk prefix len hint: 1

- id: nexmark_q2
  before:
  - create_sources
  sql: SELECT auction, price FROM bid WHERE auction = 1007 OR auction = 1020 OR auction = 2001 OR auction = 2019 OR auction = 2087;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, price] }
      └─BatchFilter { predicate: (((((auction = 1007:Int32) OR (auction = 1020:Int32)) OR (auction = 2001:Int32)) OR (auction = 2019:Int32)) OR (auction = 2087:Int32)) }
        └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, price, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, price, _row_id] }
      └─StreamFilter { predicate: (((((auction = 1007:Int32) OR (auction = 1020:Int32)) OR (auction = 2001:Int32)) OR (auction = 2019:Int32)) OR (auction = 2087:Int32)) }
        └─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, price, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [auction, price, _row_id] }
        └── StreamFilter { predicate: (((((auction = 1007:Int32) OR (auction = 1020:Int32)) OR (auction = 2001:Int32)) OR (auction = 2019:Int32)) OR (auction = 2087:Int32)) }
            └── StreamRowIdGen { row_id_index: 7 }
                └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ auction, price, _row_id ], primary key: [ $2 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 2 ], read pk prefix len hint: 1 }

- id: nexmark_q3
  before:
  - create_sources
  sql: |
    SELECT
        P.name, P.city, P.state, A.id
    FROM
        auction AS A INNER JOIN person AS P on A.seller = P.id
    WHERE
        A.category = 10 and (P.state = 'or' OR P.state = 'id' OR P.state = 'ca');
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: Inner, predicate: seller = id, output: [name, city, state, id] }
      ├─BatchExchange { order: [], dist: HashShard(seller) }
      │ └─BatchFilter { predicate: (category = 10:Int32) }
      │   └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
      └─BatchExchange { order: [], dist: HashShard(id) }
        └─BatchFilter { predicate: (((state = 'or':Varchar) OR (state = 'id':Varchar)) OR (state = 'ca':Varchar)) }
          └─BatchSource { source: person, columns: [id, name, email_address, credit_card, city, state, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [name, city, state, id, _row_id(hidden), seller(hidden), _row_id#1(hidden)], stream_key: [_row_id, _row_id#1, seller], pk_columns: [_row_id, _row_id#1, seller], pk_conflict: NoCheck }
    └─StreamHashJoin [append_only] { type: Inner, predicate: seller = id, output: [name, city, state, id, _row_id, seller, _row_id] }
      ├─StreamExchange { dist: HashShard(seller) }
      │ └─StreamFilter { predicate: (category = 10:Int32) }
      │   └─StreamRowIdGen { row_id_index: 10 }
      │     └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
      └─StreamExchange { dist: HashShard(id) }
        └─StreamFilter { predicate: (((state = 'or':Varchar) OR (state = 'id':Varchar)) OR (state = 'ca':Varchar)) }
          └─StreamRowIdGen { row_id_index: 8 }
            └─StreamSource { source: person, columns: [id, name, email_address, credit_card, city, state, date_time, extra, _row_id] }
  stream_dist_plan: |+
    Fragment 0
    StreamMaterialize { columns: [name, city, state, id, _row_id(hidden), seller(hidden), _row_id#1(hidden)], stream_key: [_row_id, _row_id#1, seller], pk_columns: [_row_id, _row_id#1, seller], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamHashJoin [append_only] { type: Inner, predicate: seller = id, output: [name, city, state, id, _row_id, seller, _row_id] } { left table: 0, right table: 2, left degree table: 1, right degree table: 3 }
        ├── StreamExchange Hash([7]) from 1
        └── StreamExchange Hash([0]) from 2

    Fragment 1
    StreamFilter { predicate: (category = 10:Int32) }
    └── StreamRowIdGen { row_id_index: 10 }
        └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 4 }

    Fragment 2
    StreamFilter { predicate: (((state = 'or':Varchar) OR (state = 'id':Varchar)) OR (state = 'ca':Varchar)) }
    └── StreamRowIdGen { row_id_index: 8 }
        └── StreamSource { source: person, columns: [id, name, email_address, credit_card, city, state, date_time, extra, _row_id] } { source state table: 5 }

    Table 0
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $7 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 7 ]
    └── read pk prefix len hint: 1

    Table 1 { columns: [ seller, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2
    ├── columns: [ id, name, email_address, credit_card, city, state, date_time, extra, _row_id ]
    ├── 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: 1

    Table 3 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 5 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ name, city, state, id, _row_id, seller, _row_id#1 ]
    ├── primary key: [ $4 ASC, $6 ASC, $5 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6 ]
    ├── distribution key: [ 5 ]
    └── read pk prefix len hint: 3

- id: nexmark_q4
  before:
  - create_sources
  sql: |
    SELECT
        Q.category,
        AVG(Q.final) as avg
    FROM (
        SELECT MAX(B.price) AS final, A.category
        FROM auction A, bid B
        WHERE A.id = B.auction AND B.date_time BETWEEN A.date_time AND A.expires
        GROUP BY A.id, A.category
    ) Q
    GROUP BY Q.category;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [category, (sum(max(price)) / count(max(price))::Decimal) as $expr1] }
      └─BatchHashAgg { group_key: [category], aggs: [sum(max(price)), count(max(price))] }
        └─BatchExchange { order: [], dist: HashShard(category) }
          └─BatchHashAgg { group_key: [id, category], aggs: [max(price)] }
            └─BatchHashJoin { type: Inner, predicate: id = auction AND (date_time >= date_time) AND (date_time <= expires), output: [id, category, price] }
              ├─BatchExchange { order: [], dist: HashShard(id) }
              │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
              └─BatchExchange { order: [], dist: HashShard(auction) }
                └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [category, avg], stream_key: [category], pk_columns: [category], pk_conflict: NoCheck }
    └─StreamProject { exprs: [category, (sum(max(price)) / count(max(price))::Decimal) as $expr1] }
      └─StreamHashAgg { group_key: [category], aggs: [sum(max(price)), count(max(price)), count] }
        └─StreamExchange { dist: HashShard(category) }
          └─StreamProject { exprs: [id, category, max(price)] }
            └─StreamHashAgg [append_only] { group_key: [id, category], aggs: [max(price), count] }
              └─StreamProject { exprs: [id, category, price, _row_id, _row_id] }
                └─StreamFilter { predicate: (date_time >= date_time) AND (date_time <= expires) }
                  └─StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: all }
                    ├─StreamExchange { dist: HashShard(id) }
                    │ └─StreamRowIdGen { row_id_index: 10 }
                    │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
                    └─StreamExchange { dist: HashShard(auction) }
                      └─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: [category, avg], stream_key: [category], pk_columns: [category], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [category, (sum(max(price)) / count(max(price))::Decimal) as $expr1] }
        └── StreamHashAgg { group_key: [category], aggs: [sum(max(price)), count(max(price)), count] }
            ├── result table: 0
            ├── state tables: []
            ├── distinct tables: []
            └── StreamExchange Hash([1]) from 1

    Fragment 1
    StreamProject { exprs: [id, category, max(price)] }
    └── StreamHashAgg [append_only] { group_key: [id, category], aggs: [max(price), count] }
        ├── result table: 1
        ├── state tables: []
        ├── distinct tables: []
        └── StreamProject { exprs: [id, category, price, _row_id, _row_id] }
            └── StreamFilter { predicate: (date_time >= date_time) AND (date_time <= expires) }
                └── StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: all }
                    ├── left table: 2
                    ├── right table: 4
                    ├── left degree table: 3
                    ├── right degree table: 5
                    ├── StreamExchange Hash([0]) from 2
                    └── StreamExchange Hash([0]) from 3

    Fragment 2
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
        └── source state table: 6

    Fragment 3
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 7 }

    Table 0
    ├── columns: [ category, sum(max(price)), count(max(price)), count ]
    ├── primary key: [ $0 ASC ]
    ├── value indices: [ 1, 2, 3 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1
    ├── columns: [ id, category, max(price), count ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2, 3 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 2

    Table 2
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 3
    ├── columns: [ id, _row_id, _degree ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

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

    Table 5
    ├── columns: [ auction, _row_id, _degree ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 6 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 7 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ category, avg ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

- id: nexmark_q5
  before:
  - create_sources
  sql: |
    SELECT AuctionBids.auction, AuctionBids.num FROM (
      SELECT
        bid.auction,
        count(*) AS num,
        window_start AS starttime
      FROM
        HOP(bid, date_time, INTERVAL '2' SECOND, INTERVAL '10' SECOND)
      GROUP BY
        window_start,
        bid.auction
    ) AS AuctionBids
    JOIN (
      SELECT
        max(CountBids.num) AS maxn,
        CountBids.starttime_c
      FROM (
        SELECT
          count(*) AS num,
          window_start AS starttime_c
        FROM HOP(bid, date_time, INTERVAL '2' SECOND, INTERVAL '10' SECOND)
        GROUP BY
          bid.auction,
          window_start
      ) AS CountBids
      GROUP BY
        CountBids.starttime_c
    ) AS MaxBids
    ON AuctionBids.starttime = MaxBids.starttime_c AND AuctionBids.num >= MaxBids.maxn;
  logical_plan: |-
    LogicalProject { exprs: [auction, count] }
    └─LogicalJoin { type: Inner, on: (window_start = window_start) AND (count >= max(count)), output: all }
      ├─LogicalProject { exprs: [auction, count, window_start] }
      │ └─LogicalAgg { group_key: [window_start, auction], aggs: [count] }
      │   └─LogicalProject { exprs: [window_start, auction] }
      │     └─LogicalHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: all }
      │       └─LogicalFilter { predicate: IsNotNull(date_time) }
      │         └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
      └─LogicalProject { exprs: [max(count), window_start] }
        └─LogicalAgg { group_key: [window_start], aggs: [max(count)] }
          └─LogicalProject { exprs: [window_start, count] }
            └─LogicalProject { exprs: [count, window_start] }
              └─LogicalAgg { group_key: [auction, window_start], aggs: [count] }
                └─LogicalProject { exprs: [auction, window_start] }
                  └─LogicalHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: all }
                    └─LogicalFilter { predicate: IsNotNull(date_time) }
                      └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: Inner, predicate: window_start = window_start AND (count >= max(count)), output: [auction, count] }
      ├─BatchExchange { order: [], dist: HashShard(window_start) }
      │ └─BatchHashAgg { group_key: [auction, window_start], aggs: [count] }
      │   └─BatchHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: [auction, window_start] }
      │     └─BatchExchange { order: [], dist: HashShard(auction) }
      │       └─BatchProject { exprs: [auction, date_time] }
      │         └─BatchFilter { predicate: IsNotNull(date_time) }
      │           └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
      └─BatchHashAgg { group_key: [window_start], aggs: [max(count)] }
        └─BatchExchange { order: [], dist: HashShard(window_start) }
          └─BatchHashAgg { group_key: [auction, window_start], aggs: [count] }
            └─BatchHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: [auction, window_start] }
              └─BatchExchange { order: [], dist: HashShard(auction) }
                └─BatchProject { exprs: [auction, date_time] }
                  └─BatchFilter { predicate: IsNotNull(date_time) }
                    └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, num, window_start(hidden), window_start#1(hidden)], stream_key: [auction, window_start, window_start#1], pk_columns: [auction, window_start, window_start#1], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, count, window_start, window_start] }
      └─StreamFilter { predicate: (count >= max(count)) }
        └─StreamHashJoin { type: Inner, predicate: window_start = window_start, output: all }
          ├─StreamExchange { dist: HashShard(window_start) }
          │ └─StreamShare { id: 7 }
          │   └─StreamHashAgg [append_only] { group_key: [auction, window_start], aggs: [count] }
          │     └─StreamExchange { dist: HashShard(auction, window_start) }
          │       └─StreamHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: [auction, window_start, _row_id] }
          │         └─StreamProject { exprs: [auction, date_time, _row_id] }
          │           └─StreamFilter { predicate: IsNotNull(date_time) }
          │             └─StreamRowIdGen { row_id_index: 7 }
          │               └─StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
          └─StreamProject { exprs: [window_start, max(count)] }
            └─StreamHashAgg { group_key: [window_start], aggs: [max(count), count] }
              └─StreamExchange { dist: HashShard(window_start) }
                └─StreamShare { id: 7 }
                  └─StreamHashAgg [append_only] { group_key: [auction, window_start], aggs: [count] }
                    └─StreamExchange { dist: HashShard(auction, window_start) }
                      └─StreamHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: [auction, window_start, _row_id] }
                        └─StreamProject { exprs: [auction, date_time, _row_id] }
                          └─StreamFilter { predicate: IsNotNull(date_time) }
                            └─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, num, window_start(hidden), window_start#1(hidden)], stream_key: [auction, window_start, window_start#1], pk_columns: [auction, window_start, window_start#1], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [auction, count, window_start, window_start] }
        └── StreamFilter { predicate: (count >= max(count)) }
            └── StreamHashJoin { type: Inner, predicate: window_start = window_start, output: all } { left table: 0, right table: 2, left degree table: 1, right degree table: 3 }
                ├── StreamExchange Hash([1]) from 1
                └── StreamProject { exprs: [window_start, max(count)] }
                    └── StreamHashAgg { group_key: [window_start], aggs: [max(count), count] } { result table: 7, state tables: [ 6 ], distinct tables: [] }
                        └── StreamExchange Hash([1]) from 4

    Fragment 1
    StreamNoOp
    └── StreamExchange NoShuffle from 2

    Fragment 2
    StreamHashAgg [append_only] { group_key: [auction, window_start], aggs: [count] } { result table: 4, state tables: [], distinct tables: [] }
    └── StreamExchange Hash([0, 1]) from 3

    Fragment 3
    StreamHopWindow { time_col: date_time, slide: 00:00:02, size: 00:00:10, output: [auction, window_start, _row_id] }
    └── StreamProject { exprs: [auction, date_time, _row_id] }
        └── StreamFilter { predicate: IsNotNull(date_time) }
            └── StreamRowIdGen { row_id_index: 7 }
                └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 5 }

    Fragment 4
    StreamNoOp
    └── StreamExchange NoShuffle from 2

    Table 0 { columns: [ auction, window_start, count ], primary key: [ $1 ASC, $0 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 1 ], read pk prefix len hint: 1 }

    Table 1 { columns: [ window_start, auction, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2 { columns: [ window_start, max(count) ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 3 { columns: [ window_start, _degree ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ auction, window_start, count ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0, 1 ], read pk prefix len hint: 2 }

    Table 5 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 6 { columns: [ window_start, count, auction ], primary key: [ $0 ASC, $1 DESC, $2 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 7 { columns: [ window_start, max(count), count ], primary key: [ $0 ASC ], value indices: [ 1, 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ auction, num, window_start, window_start#1 ], primary key: [ $0 ASC, $2 ASC, $3 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [ 2 ], read pk prefix len hint: 3 }

- id: nexmark_q6
  before:
  - create_sources
  sql: |
    SELECT
        Q.seller,
        AVG(Q.final) OVER
            (PARTITION BY Q.seller ORDER BY Q.date_time ROWS BETWEEN 10 PRECEDING AND CURRENT ROW)
        as avg
    FROM (
        SELECT MAX(B.price) AS final, A.seller, B.date_time
        FROM auction AS A, bid AS B
        WHERE A.id = B.auction and B.date_time between A.date_time and A.expires
        GROUP BY A.id, A.seller
    ) AS Q;
  planner_error: 'Invalid input syntax: column must appear in the GROUP BY clause or be used in an aggregate function'
- id: nexmark_q7
  before:
  - create_sources
  sql: |
    SELECT
      B.auction,
      B.price,
      B.bidder,
      B.date_time
    FROM
      bid B
    JOIN (
      SELECT
        MAX(price) AS maxprice,
        window_end as date_time
      FROM
        TUMBLE(bid, date_time, INTERVAL '10' SECOND)
      GROUP BY
        window_end
    ) B1 ON B.price = B1.maxprice
    WHERE
      B.date_time BETWEEN B1.date_time - INTERVAL '10' SECOND
      AND B1.date_time;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: Inner, predicate: price = max(price) AND (date_time >= $expr2) AND (date_time <= $expr1), output: [auction, price, bidder, date_time] }
      ├─BatchExchange { order: [], dist: HashShard(price) }
      │ └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
      └─BatchExchange { order: [], dist: HashShard(max(price)) }
        └─BatchProject { exprs: [max(price), $expr1, ($expr1 - '00:00:10':Interval) as $expr2] }
          └─BatchHashAgg { group_key: [$expr1], aggs: [max(price)] }
            └─BatchExchange { order: [], dist: HashShard($expr1) }
              └─BatchProject { exprs: [(TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr1, price] }
                └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, price, bidder, date_time, _row_id(hidden), $expr1(hidden)], stream_key: [_row_id, $expr1, price], pk_columns: [_row_id, $expr1, price], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, price, bidder, date_time, _row_id, $expr1] }
      └─StreamFilter { predicate: (date_time >= $expr2) AND (date_time <= $expr1) }
        └─StreamHashJoin { type: Inner, predicate: price = max(price), output: all }
          ├─StreamExchange { dist: HashShard(price) }
          │ └─StreamShare { id: 3 }
          │   └─StreamProject { exprs: [auction, bidder, price, date_time, _row_id] }
          │     └─StreamRowIdGen { row_id_index: 7 }
          │       └─StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
          └─StreamExchange { dist: HashShard(max(price)) }
            └─StreamProject { exprs: [$expr1, max(price), ($expr1 - '00:00:10':Interval) as $expr2] }
              └─StreamHashAgg [append_only] { group_key: [$expr1], aggs: [max(price), count] }
                └─StreamExchange { dist: HashShard($expr1) }
                  └─StreamProject { exprs: [(TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr1, price, _row_id] }
                    └─StreamShare { id: 3 }
                      └─StreamProject { exprs: [auction, bidder, price, date_time, _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, price, bidder, date_time, _row_id(hidden), $expr1(hidden)], stream_key: [_row_id, $expr1, price], pk_columns: [_row_id, $expr1, price], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [auction, price, bidder, date_time, _row_id, $expr1] }
        └── StreamFilter { predicate: (date_time >= $expr2) AND (date_time <= $expr1) }
            └── StreamHashJoin { type: Inner, predicate: price = max(price), output: all } { left table: 0, right table: 2, left degree table: 1, right degree table: 3 }
                ├── StreamExchange Hash([2]) from 1
                └── StreamExchange Hash([1]) from 3

    Fragment 1
    StreamNoOp
    └── StreamExchange NoShuffle from 2

    Fragment 2
    StreamProject { exprs: [auction, bidder, price, date_time, _row_id] }
    └── StreamRowIdGen { row_id_index: 7 }
        └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 4 }

    Fragment 3
    StreamProject { exprs: [$expr1, max(price), ($expr1 - '00:00:10':Interval) as $expr2] }
    └── StreamHashAgg [append_only] { group_key: [$expr1], aggs: [max(price), count] } { result table: 5, state tables: [], distinct tables: [] }
        └── StreamExchange Hash([0]) from 4

    Fragment 4
    StreamProject { exprs: [(TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr1, price, _row_id] }
    └── StreamExchange NoShuffle from 2

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

    Table 1 { columns: [ price, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2 { columns: [ $expr1, max(price), $expr2 ], primary key: [ $1 ASC, $0 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 1 ], read pk prefix len hint: 1 }

    Table 3 { columns: [ max(price), $expr1, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 5 { columns: [ $expr1, max(price), count ], primary key: [ $0 ASC ], value indices: [ 1, 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction, price, bidder, date_time, _row_id, $expr1 ]
    ├── primary key: [ $4 ASC, $5 ASC, $1 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5 ]
    ├── distribution key: [ 1 ]
    └── read pk prefix len hint: 3

- id: nexmark_q8
  before:
  - create_sources
  sql: |
    SELECT
      P.id,
      P.name,
      P.starttime
    FROM (
      SELECT
        id,
        name,
        window_start AS starttime,
        window_end AS endtime
      FROM
        TUMBLE(person, date_time, INTERVAL '10' SECOND)
      GROUP BY
        id,
        name,
        window_start,
        window_end
    ) P
    JOIN (
      SELECT
        seller,
        window_start AS starttime,
        window_end AS endtime
      FROM
        TUMBLE(auction, date_time, INTERVAL '10' SECOND)
      GROUP BY
        seller,
        window_start,
        window_end
    ) A ON P.id = A.seller
      AND P.starttime = A.starttime
      AND P.endtime = A.endtime;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: Inner, predicate: id = seller AND $expr1 = $expr3 AND $expr2 = $expr4, output: [id, name, $expr1] }
      ├─BatchExchange { order: [], dist: HashShard(id, $expr1, $expr2) }
      │ └─BatchHashAgg { group_key: [id, name, $expr1, $expr2], aggs: [] }
      │   └─BatchExchange { order: [], dist: HashShard(id, name, $expr1, $expr2) }
      │     └─BatchProject { exprs: [id, name, TumbleStart(date_time, '00:00:10':Interval) as $expr1, (TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr2] }
      │       └─BatchSource { source: person, columns: [id, name, email_address, credit_card, city, state, date_time, extra, _row_id], filter: (None, None) }
      └─BatchHashAgg { group_key: [seller, $expr3, $expr4], aggs: [] }
        └─BatchExchange { order: [], dist: HashShard(seller, $expr3, $expr4) }
          └─BatchProject { exprs: [seller, TumbleStart(date_time, '00:00:10':Interval) as $expr3, (TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr4] }
            └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [id, name, starttime, $expr2(hidden), seller(hidden), $expr3(hidden), $expr4(hidden)], stream_key: [id, name, starttime, $expr2, seller, $expr3, $expr4], pk_columns: [id, name, starttime, $expr2, seller, $expr3, $expr4], pk_conflict: NoCheck }
    └─StreamHashJoin [append_only] { type: Inner, predicate: id = seller AND $expr1 = $expr3 AND $expr2 = $expr4, output: all }
      ├─StreamExchange { dist: HashShard(id, $expr1, $expr2) }
      │ └─StreamAppendOnlyDedup { dedup_cols: [id, name, $expr1, $expr2] }
      │   └─StreamExchange { dist: HashShard(id, name, $expr1, $expr2) }
      │     └─StreamProject { exprs: [id, name, TumbleStart(date_time, '00:00:10':Interval) as $expr1, (TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr2] }
      │       └─StreamRowIdGen { row_id_index: 8 }
      │         └─StreamSource { source: person, columns: [id, name, email_address, credit_card, city, state, date_time, extra, _row_id] }
      └─StreamAppendOnlyDedup { dedup_cols: [seller, $expr3, $expr4] }
        └─StreamExchange { dist: HashShard(seller, $expr3, $expr4) }
          └─StreamProject { exprs: [seller, TumbleStart(date_time, '00:00:10':Interval) as $expr3, (TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr4] }
            └─StreamRowIdGen { row_id_index: 10 }
              └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
  stream_dist_plan: |+
    Fragment 0
    StreamMaterialize { columns: [id, name, starttime, $expr2(hidden), seller(hidden), $expr3(hidden), $expr4(hidden)], stream_key: [id, name, starttime, $expr2, seller, $expr3, $expr4], pk_columns: [id, name, starttime, $expr2, seller, $expr3, $expr4], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamHashJoin [append_only] { type: Inner, predicate: id = seller AND $expr1 = $expr3 AND $expr2 = $expr4, output: all } { left table: 0, right table: 2, left degree table: 1, right degree table: 3 }
        ├── StreamExchange Hash([0, 2, 3]) from 1
        └── StreamAppendOnlyDedup { dedup_cols: [seller, $expr3, $expr4] } { state table: 6 }
            └── StreamExchange Hash([0, 1, 2]) from 3

    Fragment 1
    StreamAppendOnlyDedup { dedup_cols: [id, name, $expr1, $expr2] } { state table: 4 }
    └── StreamExchange Hash([0, 1, 2, 3]) from 2

    Fragment 2
    StreamProject { exprs: [id, name, TumbleStart(date_time, '00:00:10':Interval) as $expr1, (TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr2] }
    └── StreamRowIdGen { row_id_index: 8 }
        └── StreamSource { source: person, columns: [id, name, email_address, credit_card, city, state, date_time, extra, _row_id] } { source state table: 5 }

    Fragment 3
    StreamProject { exprs: [seller, TumbleStart(date_time, '00:00:10':Interval) as $expr3, (TumbleStart(date_time, '00:00:10':Interval) + '00:00:10':Interval) as $expr4] }
    └── StreamRowIdGen { row_id_index: 10 }
        └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 7 }

    Table 0 { columns: [ id, name, $expr1, $expr2 ], primary key: [ $0 ASC, $2 ASC, $3 ASC, $1 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [ 0, 2, 3 ], read pk prefix len hint: 3 }

    Table 1 { columns: [ id, $expr1, $expr2, name, _degree ], primary key: [ $0 ASC, $1 ASC, $2 ASC, $3 ASC ], value indices: [ 4 ], distribution key: [ 0, 1, 2 ], read pk prefix len hint: 3 }

    Table 2 { columns: [ seller, $expr3, $expr4 ], primary key: [ $0 ASC, $1 ASC, $2 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 0, 1, 2 ], read pk prefix len hint: 3 }

    Table 3 { columns: [ seller, $expr3, $expr4, _degree ], primary key: [ $0 ASC, $1 ASC, $2 ASC ], value indices: [ 3 ], distribution key: [ 0, 1, 2 ], read pk prefix len hint: 3 }

    Table 4 { columns: [ id, name, $expr1, $expr2 ], primary key: [ $0 ASC, $1 ASC, $2 ASC, $3 ASC ], value indices: [ 0, 1, 2, 3 ], distribution key: [ 0, 1, 2, 3 ], read pk prefix len hint: 4 }

    Table 5 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 6 { columns: [ seller, $expr3, $expr4 ], primary key: [ $0 ASC, $1 ASC, $2 ASC ], value indices: [ 0, 1, 2 ], distribution key: [ 0, 1, 2 ], read pk prefix len hint: 3 }

    Table 7 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ id, name, starttime, $expr2, seller, $expr3, $expr4 ], primary key: [ $0 ASC, $1 ASC, $2 ASC, $3 ASC, $4 ASC, $5 ASC, $6 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6 ], distribution key: [ 0, 2, 3 ], read pk prefix len hint: 7 }

- id: nexmark_q9
  before:
  - create_sources
  sql: |
    SELECT
      id, item_name, description, initial_bid, reserve, date_time, expires, seller, category,
      auction, bidder, price, bid_date_time
    FROM (
      SELECT A.*, B.auction, B.bidder, B.price, B.date_time AS bid_date_time,
        ROW_NUMBER() OVER (PARTITION BY A.id ORDER BY B.price DESC, B.date_time ASC) AS rownum
      FROM auction A, bid B
      WHERE A.id = B.auction AND B.date_time BETWEEN A.date_time AND A.expires
    )
    WHERE rownum <= 1;
  logical_plan: |-
    LogicalProject { exprs: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, date_time] }
    └─LogicalFilter { predicate: (row_number <= 1:Int32) }
      └─LogicalProject { exprs: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, auction, bidder, price, date_time, row_number] }
        └─LogicalOverWindow { window_functions: [row_number() OVER(PARTITION BY id ORDER BY price DESC, date_time ASC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)] }
          └─LogicalProject { exprs: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id, auction, bidder, price, channel, url, date_time, extra, _row_id] }
            └─LogicalFilter { predicate: (id = auction) AND (date_time >= date_time) AND (date_time <= expires) }
              └─LogicalJoin { type: Inner, on: true, output: all }
                ├─LogicalSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], time_range: (Unbounded, Unbounded) }
                └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
  optimized_logical_plan_for_batch: |-
    LogicalTopN { order: [price DESC, date_time ASC], limit: 1, offset: 0, group_key: [0] }
    └─LogicalJoin { type: Inner, on: (id = auction) AND (date_time >= date_time) AND (date_time <= expires), output: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, date_time] }
      ├─LogicalSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], time_range: (Unbounded, Unbounded) }
      └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchGroupTopN { order: [price DESC, date_time ASC], limit: 1, offset: 0, group_key: [0] }
      └─BatchHashJoin { type: Inner, predicate: id = auction AND (date_time >= date_time) AND (date_time <= expires), output: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, date_time] }
        ├─BatchExchange { order: [], dist: HashShard(id) }
        │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
        └─BatchExchange { order: [], dist: HashShard(auction) }
          └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, bid_date_time, _row_id(hidden), _row_id#1(hidden)], stream_key: [id], pk_columns: [id], pk_conflict: NoCheck }
    └─StreamGroupTopN [append_only] { order: [price DESC, date_time ASC], limit: 1, offset: 0, group_key: [0] }
      └─StreamProject { exprs: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, date_time, _row_id, _row_id] }
        └─StreamFilter { predicate: (date_time >= date_time) AND (date_time <= expires) }
          └─StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: all }
            ├─StreamExchange { dist: HashShard(id) }
            │ └─StreamRowIdGen { row_id_index: 10 }
            │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
            └─StreamExchange { dist: HashShard(auction) }
              └─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: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, bid_date_time, _row_id(hidden), _row_id#1(hidden)], stream_key: [id], pk_columns: [id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamGroupTopN [append_only] { order: [price DESC, date_time ASC], limit: 1, offset: 0, group_key: [0] } { state table: 0 }
        └── StreamProject { exprs: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, date_time, _row_id, _row_id] }
            └── StreamFilter { predicate: (date_time >= date_time) AND (date_time <= expires) }
                └── StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: all } { left table: 1, right table: 3, left degree table: 2, right degree table: 4 }
                    ├── StreamExchange Hash([0]) from 1
                    └── StreamExchange Hash([0]) from 2

    Fragment 1
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 5 }

    Fragment 2
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 6 }

    Table 0
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, date_time_0, _row_id, _row_id_0 ]
    ├── primary key: [ $0 ASC, $11 DESC, $12 ASC, $13 ASC, $14 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 2 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

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

    Table 4 { columns: [ auction, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 5 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 6 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, auction, bidder, price, bid_date_time, _row_id, _row_id#1 ]
    ├── primary key: [ $0 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

- id: nexmark_q10
  before:
  - create_sources
  sql: |
    SELECT auction, bidder, price, date_time, TO_CHAR(date_time, 'YYYY-MM-DD') as date, TO_CHAR(date_time, 'HH:MI') as time FROM bid;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, bidder, price, date_time, ToChar(date_time, 'YYYY-MM-DD':Varchar) as $expr1, ToChar(date_time, 'HH:MI':Varchar) as $expr2] }
      └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, bidder, price, date_time, date, time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, bidder, price, date_time, ToChar(date_time, 'YYYY-MM-DD':Varchar) as $expr1, ToChar(date_time, 'HH:MI':Varchar) as $expr2, _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, date_time, date, time, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [auction, bidder, price, date_time, ToChar(date_time, 'YYYY-MM-DD':Varchar) as $expr1, ToChar(date_time, 'HH:MI':Varchar) as $expr2, _row_id] }
        └── StreamRowIdGen { row_id_index: 7 }
            └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction, bidder, price, date_time, date, time, _row_id ]
    ├── primary key: [ $6 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6 ]
    ├── distribution key: [ 6 ]
    └── read pk prefix len hint: 1

- id: nexmark_q11
  before:
  - create_sources
  sql: |
    SELECT
      B.bidder,
      count(*) as bid_count,
      SESSION_START(B.date_time, INTERVAL '10' SECOND) as starttime,
      SESSION_END(B.date_time, INTERVAL '10' SECOND) as endtime
    FROM bid B
    GROUP BY B.bidder, SESSION(B.date_time, INTERVAL '10' SECOND);
  binder_error: |-
    Bind error: failed to bind expression: SESSION_START(B.date_time, INTERVAL '10' SECOND)

    Caused by:
      Feature is not yet implemented: unsupported function: "session_start"
    Tracking issue: https://github.com/risingwavelabs/risingwave/issues/112
- id: nexmark_q12
  before:
  - create_sources
  sql: |
    SELECT
        B.bidder,
        count(*) as bid_count,
        TUMBLE_START(B.p_time, INTERVAL '10' SECOND) as starttime,
        TUMBLE_END(B.p_time, INTERVAL '10' SECOND) as endtime
    FROM (SELECT *, PROCTIME() as p_time FROM bid) B
    GROUP BY B.bidder, TUMBLE(B.p_time, INTERVAL '10' SECOND);
  binder_error: |-
    Bind error: failed to bind expression: PROCTIME()

    Caused by:
      Invalid input syntax: Function `PROCTIME()` is only allowed in CREATE TABLE/SOURCE. Is `NOW()` what you want?
- id: nexmark_q13
  before:
  - create_sources
  sql: |
    /* SELECT
        B.auction,
        B.bidder,
        B.price,
        B.date_time,
        S.value
    FROM (SELECT *, PROCTIME() as p_time FROM bid) B
    JOIN side_input FOR SYSTEM_TIME AS OF B.p_time AS S
    ON mod(B.auction, 10000) = S.key; */
    select 1;
  stream_error: 'Bind error: An alias must be specified for the 1st expression (counting from 1) in result relation'
- id: nexmark_q14
  before:
  - create_sources
  sql: |
    SELECT
      auction,
      bidder,
      0.908 * price as price,
      CASE
        WHEN
          extract(hour from date_time) >= 8 AND
          extract(hour from date_time) <= 18
        THEN 'dayTime'
        WHEN
          extract(hour from date_time) <= 6 OR
          extract(hour from date_time) >= 20
        THEN 'nightTime'
        ELSE 'otherTime'
      END AS bidTimeType,
      date_time,
      extra
      -- TODO: count_char is an UDF, add it back when we support similar functionality.
      -- https://github.com/nexmark/nexmark/blob/master/nexmark-flink/src/main/java/com/github/nexmark/flink/udf/CountChar.java
      -- count_char(extra, 'c') AS c_counts
    FROM bid
    WHERE 0.908 * price > 1000000 AND 0.908 * price < 50000000;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, bidder, (0.908:Decimal * price::Decimal) as $expr1, Case(((Extract('HOUR':Varchar, date_time) >= 8:Decimal) AND (Extract('HOUR':Varchar, date_time) <= 18:Decimal)), 'dayTime':Varchar, ((Extract('HOUR':Varchar, date_time) <= 6:Decimal) OR (Extract('HOUR':Varchar, date_time) >= 20:Decimal)), 'nightTime':Varchar, 'otherTime':Varchar) as $expr2, date_time, extra] }
      └─BatchFilter { predicate: ((0.908:Decimal * price::Decimal) > 1000000:Decimal) AND ((0.908:Decimal * price::Decimal) < 50000000:Decimal) }
        └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, bidder, price, bidtimetype, date_time, extra, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, bidder, (0.908:Decimal * price::Decimal) as $expr1, Case(((Extract('HOUR':Varchar, date_time) >= 8:Decimal) AND (Extract('HOUR':Varchar, date_time) <= 18:Decimal)), 'dayTime':Varchar, ((Extract('HOUR':Varchar, date_time) <= 6:Decimal) OR (Extract('HOUR':Varchar, date_time) >= 20:Decimal)), 'nightTime':Varchar, 'otherTime':Varchar) as $expr2, date_time, extra, _row_id] }
      └─StreamFilter { predicate: ((0.908:Decimal * price::Decimal) > 1000000:Decimal) AND ((0.908:Decimal * price::Decimal) < 50000000:Decimal) }
        └─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, bidtimetype, date_time, extra, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [auction, bidder, (0.908:Decimal * price::Decimal) as $expr1, Case(((Extract('HOUR':Varchar, date_time) >= 8:Decimal) AND (Extract('HOUR':Varchar, date_time) <= 18:Decimal)), 'dayTime':Varchar, ((Extract('HOUR':Varchar, date_time) <= 6:Decimal) OR (Extract('HOUR':Varchar, date_time) >= 20:Decimal)), 'nightTime':Varchar, 'otherTime':Varchar) as $expr2, date_time, extra, _row_id] }
        └── StreamFilter { predicate: ((0.908:Decimal * price::Decimal) > 1000000:Decimal) AND ((0.908:Decimal * price::Decimal) < 50000000:Decimal) }
            └── StreamRowIdGen { row_id_index: 7 }
                └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

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

- id: nexmark_q15
  before:
  - create_sources
  sql: |
    SELECT
        TO_CHAR(date_time, 'yyyy-MM-dd') as day,
        count(*) AS total_bids,
        count(*) filter (where price < 10000) AS rank1_bids,
        count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
        count(*) filter (where price >= 1000000) AS rank3_bids,
        count(distinct bidder) AS total_bidders,
        count(distinct bidder) filter (where price < 10000) AS rank1_bidders,
        count(distinct bidder) filter (where price >= 10000 and price < 1000000) AS rank2_bidders,
        count(distinct bidder) filter (where price >= 1000000) AS rank3_bidders,
        count(distinct auction) AS total_auctions,
        count(distinct auction) filter (where price < 10000) AS rank1_auctions,
        count(distinct auction) filter (where price >= 10000 and price < 1000000) AS rank2_auctions,
        count(distinct auction) filter (where price >= 1000000) AS rank3_auctions
    FROM bid
    GROUP BY to_char(date_time, 'yyyy-MM-dd');
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashAgg { group_key: [$expr1], aggs: [sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64))] }
      └─BatchExchange { order: [], dist: HashShard($expr1) }
        └─BatchHashAgg { group_key: [$expr1, bidder, auction, flag], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] }
          └─BatchExchange { order: [], dist: HashShard($expr1, bidder, auction, flag) }
            └─BatchExpand { column_subsets: [[$expr1], [$expr1, bidder], [$expr1, auction]] }
              └─BatchProject { exprs: [ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, price, bidder, auction] }
                └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [day, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [day], pk_columns: [day], pk_conflict: NoCheck }
    └─StreamHashAgg [append_only] { group_key: [$expr1], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count(distinct bidder), count(distinct bidder) filter((price < 10000:Int32)), count(distinct bidder) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct bidder) filter((price >= 1000000:Int32)), count(distinct auction), count(distinct auction) filter((price < 10000:Int32)), count(distinct auction) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct auction) filter((price >= 1000000:Int32))] }
      └─StreamExchange { dist: HashShard($expr1) }
        └─StreamProject { exprs: [ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, price, bidder, auction, _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: [day, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [day], pk_columns: [day], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamHashAgg [append_only] { group_key: [$expr1], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count(distinct bidder), count(distinct bidder) filter((price < 10000:Int32)), count(distinct bidder) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct bidder) filter((price >= 1000000:Int32)), count(distinct auction), count(distinct auction) filter((price < 10000:Int32)), count(distinct auction) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct auction) filter((price >= 1000000:Int32))] }
        ├── result table: 0
        ├── state tables: []
        ├── distinct tables: [ (distinct key: bidder, table id: 1), (distinct key: auction, table id: 2) ]
        └── StreamExchange Hash([0]) from 1

    Fragment 1
    StreamProject { exprs: [ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, price, bidder, auction, _row_id] }
    └── StreamRowIdGen { row_id_index: 7 }
        └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 3 }

    Table 0
    ├── columns: [ $expr1, count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count(distinct bidder), count(distinct bidder) filter((price < 10000:Int32)), count(distinct bidder) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct bidder) filter((price >= 1000000:Int32)), count(distinct auction), count(distinct auction) filter((price < 10000:Int32)), count(distinct auction) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct auction) filter((price >= 1000000:Int32)) ]
    ├── primary key: [ $0 ASC ]
    ├── value indices: [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1 { columns: [ $expr1, bidder, count_for_agg_call_4, count_for_agg_call_5, count_for_agg_call_6, count_for_agg_call_7 ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2, 3, 4, 5 ], distribution key: [ 0 ], read pk prefix len hint: 2 }

    Table 2 { columns: [ $expr1, auction, count_for_agg_call_8, count_for_agg_call_9, count_for_agg_call_10, count_for_agg_call_11 ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2, 3, 4, 5 ], distribution key: [ 0 ], read pk prefix len hint: 2 }

    Table 3 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ day, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions ], primary key: [ $0 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

- id: nexmark_q15_split_distinct_agg
  before:
  - create_sources
  sql: |
    SELECT
        TO_CHAR(date_time, 'yyyy-MM-dd') as day,
        count(*) AS total_bids,
        count(*) filter (where price < 10000) AS rank1_bids,
        count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
        count(*) filter (where price >= 1000000) AS rank3_bids,
        count(distinct bidder) AS total_bidders,
        count(distinct bidder) filter (where price < 10000) AS rank1_bidders,
        count(distinct bidder) filter (where price >= 10000 and price < 1000000) AS rank2_bidders,
        count(distinct bidder) filter (where price >= 1000000) AS rank3_bidders,
        count(distinct auction) AS total_auctions,
        count(distinct auction) filter (where price < 10000) AS rank1_auctions,
        count(distinct auction) filter (where price >= 10000 and price < 1000000) AS rank2_auctions,
        count(distinct auction) filter (where price >= 1000000) AS rank3_auctions
    FROM bid
    GROUP BY to_char(date_time, 'yyyy-MM-dd');
  stream_plan: |-
    StreamMaterialize { columns: [day, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [day], pk_columns: [day], pk_conflict: NoCheck }
    └─StreamProject { exprs: [$expr1, sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64))] }
      └─StreamHashAgg { group_key: [$expr1], aggs: [sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count] }
        └─StreamExchange { dist: HashShard($expr1) }
          └─StreamHashAgg [append_only] { group_key: [$expr1, bidder, auction, flag], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] }
            └─StreamExchange { dist: HashShard($expr1, bidder, auction, flag) }
              └─StreamExpand { column_subsets: [[$expr1], [$expr1, bidder], [$expr1, auction]] }
                └─StreamProject { exprs: [ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, price, bidder, auction, _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: [day, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [day], pk_columns: [day], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [$expr1, sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64))] }
        └── StreamHashAgg { group_key: [$expr1], aggs: [sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count] }
            ├── result table: 0
            ├── state tables: []
            ├── distinct tables: []
            └── StreamExchange Hash([0]) from 1

    Fragment 1
    StreamHashAgg [append_only] { group_key: [$expr1, bidder, auction, flag], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] } { result table: 1, state tables: [], distinct tables: [] }
    └── StreamExchange Hash([0, 2, 3, 10]) from 2

    Fragment 2
    StreamExpand { column_subsets: [[$expr1], [$expr1, bidder], [$expr1, auction]] }
    └── StreamProject { exprs: [ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, price, bidder, auction, _row_id] }
        └── StreamRowIdGen { row_id_index: 7 }
            └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 2 }

    Table 0
    ├── columns: [ $expr1, sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count ]
    ├── primary key: [ $0 ASC ]
    ├── value indices: [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1 { columns: [ $expr1, bidder, auction, flag, count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32))_0, count filter((price >= 10000:Int32) AND (price < 1000000:Int32))_0, count filter((price >= 1000000:Int32))_0, count filter((price < 10000:Int32))_1, count filter((price >= 10000:Int32) AND (price < 1000000:Int32))_1, count filter((price >= 1000000:Int32))_1 ], primary key: [ $0 ASC, $1 ASC, $2 ASC, $3 ASC ], value indices: [ 4, 5, 6, 7, 8, 9, 10, 11, 12, 13 ], distribution key: [ 0, 1, 2, 3 ], read pk prefix len hint: 4 }

    Table 2 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ day, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions ], primary key: [ $0 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

  with_config_map:
    RW_FORCE_SPLIT_DISTINCT_AGG: 'true'
- id: nexmark_q16
  before:
  - create_sources
  sql: |
    SELECT
      channel,
      to_char(date_time, 'yyyy-MM-dd') AS day,
      max(to_char(date_time, 'HH:mm')) AS minute,
      count(*) AS total_bids,
      count(*) filter (where price < 10000) AS rank1_bids,
      count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
      count(*) filter (where price >= 1000000) AS rank3_bids,
      count(distinct bidder) AS total_bidders,
      count(distinct bidder) filter (where price < 10000) AS rank1_bidders,
      count(distinct bidder) filter (where price >= 10000 and price < 1000000) AS rank2_bidders,
      count(distinct bidder) filter (where price >= 1000000) AS rank3_bidders,
      count(distinct auction) AS total_auctions,
      count(distinct auction) filter (where price < 10000) AS rank1_auctions,
      count(distinct auction) filter (where price >= 10000 and price < 1000000) AS rank2_auctions,
      count(distinct auction) filter (where price >= 1000000) AS rank3_auctions
    FROM bid
    GROUP BY channel, to_char(date_time, 'yyyy-MM-dd');
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashAgg { group_key: [channel, $expr1], aggs: [max(max($expr2)) filter((flag = 0:Int64)), sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64))] }
      └─BatchExchange { order: [], dist: HashShard(channel, $expr1) }
        └─BatchHashAgg { group_key: [channel, $expr1, bidder, auction, flag], aggs: [max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] }
          └─BatchExchange { order: [], dist: HashShard(channel, $expr1, bidder, auction, flag) }
            └─BatchExpand { column_subsets: [[channel, $expr1, $expr2], [channel, $expr1, bidder], [channel, $expr1, auction]] }
              └─BatchProject { exprs: [channel, ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, ToChar(date_time, 'HH:mm':Varchar) as $expr2, price, bidder, auction] }
                └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [channel, day, minute, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [channel, day], pk_columns: [channel, day], pk_conflict: NoCheck }
    └─StreamHashAgg [append_only] { group_key: [channel, $expr1], aggs: [max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count(distinct bidder), count(distinct bidder) filter((price < 10000:Int32)), count(distinct bidder) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct bidder) filter((price >= 1000000:Int32)), count(distinct auction), count(distinct auction) filter((price < 10000:Int32)), count(distinct auction) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct auction) filter((price >= 1000000:Int32))] }
      └─StreamExchange { dist: HashShard(channel, $expr1) }
        └─StreamProject { exprs: [channel, ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, ToChar(date_time, 'HH:mm':Varchar) as $expr2, price, bidder, auction, _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: [channel, day, minute, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [channel, day], pk_columns: [channel, day], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamHashAgg [append_only] { group_key: [channel, $expr1], aggs: [max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count(distinct bidder), count(distinct bidder) filter((price < 10000:Int32)), count(distinct bidder) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct bidder) filter((price >= 1000000:Int32)), count(distinct auction), count(distinct auction) filter((price < 10000:Int32)), count(distinct auction) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct auction) filter((price >= 1000000:Int32))] }
        ├── result table: 0
        ├── state tables: []
        ├── distinct tables: [ (distinct key: bidder, table id: 1), (distinct key: auction, table id: 2) ]
        └── StreamExchange Hash([0, 1]) from 1

    Fragment 1
    StreamProject { exprs: [channel, ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, ToChar(date_time, 'HH:mm':Varchar) as $expr2, price, bidder, auction, _row_id] }
    └── StreamRowIdGen { row_id_index: 7 }
        └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 3 }

    Table 0
    ├── columns: [ channel, $expr1, max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count(distinct bidder), count(distinct bidder) filter((price < 10000:Int32)), count(distinct bidder) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct bidder) filter((price >= 1000000:Int32)), count(distinct auction), count(distinct auction) filter((price < 10000:Int32)), count(distinct auction) filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count(distinct auction) filter((price >= 1000000:Int32)) ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 ]
    ├── distribution key: [ 0, 1 ]
    └── read pk prefix len hint: 2

    Table 1 { columns: [ channel, $expr1, bidder, count_for_agg_call_5, count_for_agg_call_6, count_for_agg_call_7, count_for_agg_call_8 ], primary key: [ $0 ASC, $1 ASC, $2 ASC ], value indices: [ 3, 4, 5, 6 ], distribution key: [ 0, 1 ], read pk prefix len hint: 3 }

    Table 2 { columns: [ channel, $expr1, auction, count_for_agg_call_9, count_for_agg_call_10, count_for_agg_call_11, count_for_agg_call_12 ], primary key: [ $0 ASC, $1 ASC, $2 ASC ], value indices: [ 3, 4, 5, 6 ], distribution key: [ 0, 1 ], read pk prefix len hint: 3 }

    Table 3 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ channel, day, minute, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 ], distribution key: [ 0, 1 ], read pk prefix len hint: 2 }

- id: nexmark_q16
  before:
  - create_sources
  sql: |
    SELECT
      channel,
      to_char(date_time, 'yyyy-MM-dd') AS day,
      max(to_char(date_time, 'HH:mm')) AS minute,
      count(*) AS total_bids,
      count(*) filter (where price < 10000) AS rank1_bids,
      count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
      count(*) filter (where price >= 1000000) AS rank3_bids,
      count(distinct bidder) AS total_bidders,
      count(distinct bidder) filter (where price < 10000) AS rank1_bidders,
      count(distinct bidder) filter (where price >= 10000 and price < 1000000) AS rank2_bidders,
      count(distinct bidder) filter (where price >= 1000000) AS rank3_bidders,
      count(distinct auction) AS total_auctions,
      count(distinct auction) filter (where price < 10000) AS rank1_auctions,
      count(distinct auction) filter (where price >= 10000 and price < 1000000) AS rank2_auctions,
      count(distinct auction) filter (where price >= 1000000) AS rank3_auctions
    FROM bid
    GROUP BY channel, to_char(date_time, 'yyyy-MM-dd');
  stream_plan: |-
    StreamMaterialize { columns: [channel, day, minute, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [channel, day], pk_columns: [channel, day], pk_conflict: NoCheck }
    └─StreamProject { exprs: [channel, $expr1, max(max($expr2)) filter((flag = 0:Int64)), sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64))] }
      └─StreamHashAgg { group_key: [channel, $expr1], aggs: [max(max($expr2)) filter((flag = 0:Int64)), sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count] }
        └─StreamExchange { dist: HashShard(channel, $expr1) }
          └─StreamHashAgg [append_only] { group_key: [channel, $expr1, bidder, auction, flag], aggs: [max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] }
            └─StreamExchange { dist: HashShard(channel, $expr1, bidder, auction, flag) }
              └─StreamExpand { column_subsets: [[channel, $expr1, $expr2], [channel, $expr1, bidder], [channel, $expr1, auction]] }
                └─StreamProject { exprs: [channel, ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, ToChar(date_time, 'HH:mm':Varchar) as $expr2, price, bidder, auction, _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: [channel, day, minute, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions], stream_key: [channel, day], pk_columns: [channel, day], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [channel, $expr1, max(max($expr2)) filter((flag = 0:Int64)), sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64))] }
        └── StreamHashAgg { group_key: [channel, $expr1], aggs: [max(max($expr2)) filter((flag = 0:Int64)), sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count] }
            ├── result table: 1
            ├── state tables: [ 0 ]
            ├── distinct tables: []
            └── StreamExchange Hash([0, 1]) from 1

    Fragment 1
    StreamHashAgg [append_only] { group_key: [channel, $expr1, bidder, auction, flag], aggs: [max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32))] } { result table: 2, state tables: [], distinct tables: [] }
    └── StreamExchange Hash([0, 1, 4, 5, 14]) from 2

    Fragment 2
    StreamExpand { column_subsets: [[channel, $expr1, $expr2], [channel, $expr1, bidder], [channel, $expr1, auction]] }
    └── StreamProject { exprs: [channel, ToChar(date_time, 'yyyy-MM-dd':Varchar) as $expr1, ToChar(date_time, 'HH:mm':Varchar) as $expr2, price, bidder, auction, _row_id] }
        └── StreamRowIdGen { row_id_index: 7 }
            └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 3 }

    Table 0 { columns: [ channel, $expr1, max($expr2), bidder, auction, flag ], primary key: [ $0 ASC, $1 ASC, $2 DESC, $3 ASC, $4 ASC, $5 ASC ], value indices: [ 0, 1, 2, 3, 4, 5 ], distribution key: [ 0, 1 ], read pk prefix len hint: 2 }

    Table 1
    ├── columns: [ channel, $expr1, max(max($expr2)) filter((flag = 0:Int64)), sum0(count) filter((flag = 0:Int64)), sum0(count filter((price < 10000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 10000:Int32) AND (price < 1000000:Int32))) filter((flag = 0:Int64)), sum0(count filter((price >= 1000000:Int32))) filter((flag = 0:Int64)), count(bidder) filter((flag = 1:Int64)), count(bidder) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(bidder) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 1:Int64)), count(auction) filter((flag = 2:Int64)), count(auction) filter((count filter((price < 10000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 10000:Int32) AND (price < 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count(auction) filter((count filter((price >= 1000000:Int32)) > 0:Int64) AND (flag = 2:Int64)), count ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 ]
    ├── distribution key: [ 0, 1 ]
    └── read pk prefix len hint: 2

    Table 2 { columns: [ channel, $expr1, bidder, auction, flag, max($expr2), count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), count filter((price < 10000:Int32))_0, count filter((price >= 10000:Int32) AND (price < 1000000:Int32))_0, count filter((price >= 1000000:Int32))_0, count filter((price < 10000:Int32))_1, count filter((price >= 10000:Int32) AND (price < 1000000:Int32))_1, count filter((price >= 1000000:Int32))_1 ], primary key: [ $0 ASC, $1 ASC, $2 ASC, $3 ASC, $4 ASC ], value indices: [ 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 ], distribution key: [ 0, 1, 2, 3, 4 ], read pk prefix len hint: 5 }

    Table 3 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ channel, day, minute, total_bids, rank1_bids, rank2_bids, rank3_bids, total_bidders, rank1_bidders, rank2_bidders, rank3_bidders, total_auctions, rank1_auctions, rank2_auctions, rank3_auctions ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14 ], distribution key: [ 0, 1 ], read pk prefix len hint: 2 }

  with_config_map:
    RW_FORCE_SPLIT_DISTINCT_AGG: 'true'
- id: nexmark_q17
  before:
  - create_sources
  sql: |
    SELECT
        auction,
        to_char(date_time, 'YYYY-MM-DD') AS day,
        count(*) AS total_bids,
        count(*) filter (where price < 10000) AS rank1_bids,
        count(*) filter (where price >= 10000 and price < 1000000) AS rank2_bids,
        count(*) filter (where price >= 1000000) AS rank3_bids,
        min(price) AS min_price,
        max(price) AS max_price,
        avg(price) AS avg_price,
        sum(price) AS sum_price
    FROM bid
    GROUP BY auction, to_char(date_time, 'YYYY-MM-DD');
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, $expr1, count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), (sum(price) / count(price)::Decimal) as $expr2, sum(price)] }
      └─BatchHashAgg { group_key: [auction, $expr1], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), sum(price), count(price)] }
        └─BatchExchange { order: [], dist: HashShard(auction, $expr1) }
          └─BatchProject { exprs: [auction, ToChar(date_time, 'YYYY-MM-DD':Varchar) as $expr1, price] }
            └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, day, total_bids, rank1_bids, rank2_bids, rank3_bids, min_price, max_price, avg_price, sum_price], stream_key: [auction, day], pk_columns: [auction, day], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, $expr1, count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), (sum(price) / count(price)::Decimal) as $expr2, sum(price)] }
      └─StreamHashAgg [append_only] { group_key: [auction, $expr1], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), sum(price), count(price)] }
        └─StreamExchange { dist: HashShard(auction, $expr1) }
          └─StreamProject { exprs: [auction, ToChar(date_time, 'YYYY-MM-DD':Varchar) as $expr1, price, _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, day, total_bids, rank1_bids, rank2_bids, rank3_bids, min_price, max_price, avg_price, sum_price], stream_key: [auction, day], pk_columns: [auction, day], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [auction, $expr1, count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), (sum(price) / count(price)::Decimal) as $expr2, sum(price)] }
        └── StreamHashAgg [append_only] { group_key: [auction, $expr1], aggs: [count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), sum(price), count(price)] }
            ├── result table: 0
            ├── state tables: []
            ├── distinct tables: []
            └── StreamExchange Hash([0, 1]) from 1

    Fragment 1
    StreamProject { exprs: [auction, ToChar(date_time, 'YYYY-MM-DD':Varchar) as $expr1, price, _row_id] }
    └── StreamRowIdGen { row_id_index: 7 }
        └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 1 }

    Table 0
    ├── columns: [ auction, $expr1, count, count filter((price < 10000:Int32)), count filter((price >= 10000:Int32) AND (price < 1000000:Int32)), count filter((price >= 1000000:Int32)), min(price), max(price), sum(price), count(price) ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2, 3, 4, 5, 6, 7, 8, 9 ]
    ├── distribution key: [ 0, 1 ]
    └── read pk prefix len hint: 2

    Table 1 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ auction, day, total_bids, rank1_bids, rank2_bids, rank3_bids, min_price, max_price, avg_price, sum_price ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ], distribution key: [ 0, 1 ], read pk prefix len hint: 2 }

- id: nexmark_q18
  before:
  - create_sources
  sql: |
    SELECT auction, bidder, price, channel, url, date_time, extra
    FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY bidder, auction ORDER BY date_time DESC) AS rank_number
          FROM bid)
    WHERE rank_number <= 1;
  logical_plan: |-
    LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra] }
    └─LogicalFilter { predicate: (row_number <= 1:Int32) }
      └─LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, row_number] }
        └─LogicalOverWindow { window_functions: [row_number() OVER(PARTITION BY bidder, auction ORDER BY date_time DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)] }
          └─LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
            └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchGroupTopN { order: [date_time DESC], limit: 1, offset: 0, group_key: [1, 0] }
      └─BatchExchange { order: [], dist: HashShard(bidder, auction) }
        └─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: [bidder, auction], pk_columns: [bidder, auction], pk_conflict: NoCheck }
    └─StreamGroupTopN [append_only] { order: [date_time DESC], limit: 1, offset: 0, group_key: [1, 0] }
      └─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: [bidder, auction], pk_columns: [bidder, auction], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamGroupTopN [append_only] { order: [date_time DESC], limit: 1, offset: 0, group_key: [1, 0] } { state table: 0 }
        └── StreamExchange Hash([1, 0]) from 1

    Fragment 1
    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] } { source state table: 1 }

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

    Table 1 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

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

- id: nexmark_q18_rank
  before:
  - create_sources
  sql: |
    SELECT auction, bidder, price, channel, url, date_time, extra
    FROM (SELECT *, RANK() OVER (PARTITION BY bidder, auction ORDER BY date_time DESC) AS rank_number
          FROM bid)
    WHERE rank_number <= 1;
  logical_plan: |-
    LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra] }
    └─LogicalFilter { predicate: (rank <= 1:Int32) }
      └─LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, rank] }
        └─LogicalOverWindow { window_functions: [rank() OVER(PARTITION BY bidder, auction ORDER BY date_time DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)] }
          └─LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
            └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchGroupTopN { order: [date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [1, 0] }
      └─BatchExchange { order: [], dist: HashShard(bidder, auction) }
        └─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], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamExchange { dist: HashShard(_row_id) }
      └─StreamGroupTopN [append_only] { order: [date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [1, 0] }
        └─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], pk_columns: [_row_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamExchange Hash([7]) from 1

    Fragment 1
    StreamGroupTopN [append_only] { order: [date_time DESC], limit: 1, offset: 0, with_ties: true, group_key: [1, 0] } { state table: 0 }
    └── StreamExchange Hash([1, 0]) from 2

    Fragment 2
    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] } { source state table: 1 }

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

    Table 1 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

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

- id: nexmark_q19
  before:
  - create_sources
  sql: |
    SELECT * FROM
    (SELECT *, ROW_NUMBER() OVER (PARTITION BY auction ORDER BY price DESC) AS rank_number FROM bid)
    WHERE rank_number <= 10;
  logical_plan: |-
    LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, row_number] }
    └─LogicalFilter { predicate: (row_number <= 10:Int32) }
      └─LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, row_number] }
        └─LogicalOverWindow { window_functions: [row_number() OVER(PARTITION BY auction ORDER BY price DESC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)] }
          └─LogicalProject { exprs: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
            └─LogicalSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], time_range: (Unbounded, Unbounded) }
- id: nexmark_q20
  before:
  - create_sources
  sql: |
    SELECT
        auction, bidder, price, channel, url, B.date_time as date_timeB,
        item_name, description, initial_bid, reserve, A.date_time as date_timeA, expires, seller, category
    FROM
        bid B INNER JOIN auction A on B.auction = A.id
    WHERE A.category = 10;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: Inner, predicate: auction = id, output: [auction, bidder, price, channel, url, date_time, item_name, description, initial_bid, reserve, date_time, expires, seller, category] }
      ├─BatchExchange { order: [], dist: HashShard(auction) }
      │ └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
      └─BatchExchange { order: [], dist: HashShard(id) }
        └─BatchFilter { predicate: (category = 10:Int32) }
          └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction, bidder, price, channel, url, date_timeb, item_name, description, initial_bid, reserve, date_timea, expires, seller, category, _row_id(hidden), _row_id#1(hidden)], stream_key: [_row_id, _row_id#1, auction], pk_columns: [_row_id, _row_id#1, auction], pk_conflict: NoCheck }
    └─StreamHashJoin [append_only] { type: Inner, predicate: auction = id, output: [auction, bidder, price, channel, url, date_time, item_name, description, initial_bid, reserve, date_time, expires, seller, category, _row_id, _row_id] }
      ├─StreamExchange { dist: HashShard(auction) }
      │ └─StreamRowIdGen { row_id_index: 7 }
      │   └─StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
      └─StreamExchange { dist: HashShard(id) }
        └─StreamFilter { predicate: (category = 10:Int32) }
          └─StreamRowIdGen { row_id_index: 10 }
            └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
  stream_dist_plan: |+
    Fragment 0
    StreamMaterialize { columns: [auction, bidder, price, channel, url, date_timeb, item_name, description, initial_bid, reserve, date_timea, expires, seller, category, _row_id(hidden), _row_id#1(hidden)], stream_key: [_row_id, _row_id#1, auction], pk_columns: [_row_id, _row_id#1, auction], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamHashJoin [append_only] { type: Inner, predicate: auction = id, output: [auction, bidder, price, channel, url, date_time, item_name, description, initial_bid, reserve, date_time, expires, seller, category, _row_id, _row_id] }
        ├── left table: 0
        ├── right table: 2
        ├── left degree table: 1
        ├── right degree table: 3
        ├── StreamExchange Hash([0]) from 1
        └── StreamExchange Hash([0]) from 2

    Fragment 1
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 4 }

    Fragment 2
    StreamFilter { predicate: (category = 10:Int32) }
    └── StreamRowIdGen { row_id_index: 10 }
        └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 5 }

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

    Table 1 { columns: [ auction, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2 { columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ], primary key: [ $0 ASC, $10 ASC ], value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 3 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 5 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction, bidder, price, channel, url, date_timeb, item_name, description, initial_bid, reserve, date_timea, expires, seller, category, _row_id, _row_id#1 ]
    ├── primary key: [ $14 ASC, $15 ASC, $0 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 3

- id: nexmark_q21
  before:
  - create_sources
  sql: |
    SELECT
        auction, bidder, price, channel,
        CASE
            WHEN lower(channel) = 'apple' THEN '0'
            WHEN lower(channel) = 'google' THEN '1'
            WHEN lower(channel) = 'facebook' THEN '2'
            WHEN lower(channel) = 'baidu' THEN '3'
            ELSE REGEXP_MATCH(url, '(&|^)channel_id=([^&]*)')[2]
            END
        AS channel_id FROM bid
        where REGEXP_MATCH(url, '(&|^)channel_id=([^&]*)')[2] is not null or
              lower(channel) in ('apple', 'google', 'facebook', 'baidu');
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, bidder, price, channel, Case((Lower(channel) = 'apple':Varchar), '0':Varchar, (Lower(channel) = 'google':Varchar), '1':Varchar, (Lower(channel) = 'facebook':Varchar), '2':Varchar, (Lower(channel) = 'baidu':Varchar), '3':Varchar, ArrayAccess(RegexpMatch(url, '(&|^)channel_id=([^&]*)':Varchar), 2:Int32)) as $expr1] }
      └─BatchFilter { predicate: (IsNotNull(ArrayAccess(RegexpMatch(url, '(&|^)channel_id=([^&]*)':Varchar), 2:Int32)) OR In(Lower(channel), 'apple':Varchar, 'google':Varchar, 'facebook':Varchar, 'baidu':Varchar)) }
        └─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, channel_id, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, bidder, price, channel, Case((Lower(channel) = 'apple':Varchar), '0':Varchar, (Lower(channel) = 'google':Varchar), '1':Varchar, (Lower(channel) = 'facebook':Varchar), '2':Varchar, (Lower(channel) = 'baidu':Varchar), '3':Varchar, ArrayAccess(RegexpMatch(url, '(&|^)channel_id=([^&]*)':Varchar), 2:Int32)) as $expr1, _row_id] }
      └─StreamFilter { predicate: (IsNotNull(ArrayAccess(RegexpMatch(url, '(&|^)channel_id=([^&]*)':Varchar), 2:Int32)) OR In(Lower(channel), 'apple':Varchar, 'google':Varchar, 'facebook':Varchar, 'baidu':Varchar)) }
        └─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, channel_id, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [auction, bidder, price, channel, Case((Lower(channel) = 'apple':Varchar), '0':Varchar, (Lower(channel) = 'google':Varchar), '1':Varchar, (Lower(channel) = 'facebook':Varchar), '2':Varchar, (Lower(channel) = 'baidu':Varchar), '3':Varchar, ArrayAccess(RegexpMatch(url, '(&|^)channel_id=([^&]*)':Varchar), 2:Int32)) as $expr1, _row_id] }
        └── StreamFilter { predicate: (IsNotNull(ArrayAccess(RegexpMatch(url, '(&|^)channel_id=([^&]*)':Varchar), 2:Int32)) OR In(Lower(channel), 'apple':Varchar, 'google':Varchar, 'facebook':Varchar, 'baidu':Varchar)) }
            └── StreamRowIdGen { row_id_index: 7 }
                └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

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

- id: nexmark_q22
  before:
  - create_sources
  sql: |
    SELECT
        auction, bidder, price, channel,
        SPLIT_PART(url, '/', 4) as dir1,
        SPLIT_PART(url, '/', 5) as dir2,
        SPLIT_PART(url, '/', 6) as dir3 FROM bid;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchProject { exprs: [auction, bidder, price, channel, SplitPart(url, '/':Varchar, 4:Int32) as $expr1, SplitPart(url, '/':Varchar, 5:Int32) as $expr2, SplitPart(url, '/':Varchar, 6:Int32) as $expr3] }
      └─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, dir1, dir2, dir3, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck }
    └─StreamProject { exprs: [auction, bidder, price, channel, SplitPart(url, '/':Varchar, 4:Int32) as $expr1, SplitPart(url, '/':Varchar, 5:Int32) as $expr2, SplitPart(url, '/':Varchar, 6:Int32) as $expr3, _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, dir1, dir2, dir3, _row_id(hidden)], stream_key: [_row_id], pk_columns: [_row_id], pk_conflict: NoCheck } { materialized table: 4294967294 }
    └── StreamProject { exprs: [auction, bidder, price, channel, SplitPart(url, '/':Varchar, 4:Int32) as $expr1, SplitPart(url, '/':Varchar, 5:Int32) as $expr2, SplitPart(url, '/':Varchar, 6:Int32) as $expr3, _row_id] }
        └── StreamRowIdGen { row_id_index: 7 }
            └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 0 }

    Table 0 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

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

- id: nexmark_q101
  before:
  - create_sources
  sql: |
    -- A self-made query that covers outer join.
    --
    -- Monitor ongoing auctions and track the current highest bid for each one in real-time. If
    -- the auction has no bids, the highest bid will be NULL.
    SELECT
        a.id AS auction_id,
        a.item_name AS auction_item_name,
        b.max_price AS current_highest_bid
    FROM auction a
    LEFT OUTER JOIN (
        SELECT
            b1.auction,
            MAX(b1.price) max_price
        FROM bid b1
        GROUP BY b1.auction
    ) b ON a.id = b.auction;
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: LeftOuter, predicate: id = auction, output: [id, item_name, max(price)] }
      ├─BatchExchange { order: [], dist: HashShard(id) }
      │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
      └─BatchHashAgg { group_key: [auction], aggs: [max(price)] }
        └─BatchExchange { order: [], dist: HashShard(auction) }
          └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction_id, auction_item_name, current_highest_bid, _row_id(hidden), auction(hidden)], stream_key: [_row_id, auction, auction_id], pk_columns: [_row_id, auction, auction_id], pk_conflict: NoCheck }
    └─StreamHashJoin { type: LeftOuter, predicate: id = auction, output: [id, item_name, max(price), _row_id, auction] }
      ├─StreamExchange { dist: HashShard(id) }
      │ └─StreamRowIdGen { row_id_index: 10 }
      │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
      └─StreamProject { exprs: [auction, max(price)] }
        └─StreamHashAgg [append_only] { group_key: [auction], aggs: [max(price), count] }
          └─StreamExchange { dist: HashShard(auction) }
            └─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_id, auction_item_name, current_highest_bid, _row_id(hidden), auction(hidden)], stream_key: [_row_id, auction, auction_id], pk_columns: [_row_id, auction, auction_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamHashJoin { type: LeftOuter, predicate: id = auction, output: [id, item_name, max(price), _row_id, auction] } { left table: 0, right table: 2, left degree table: 1, right degree table: 3 }
        ├── StreamExchange Hash([0]) from 1
        └── StreamProject { exprs: [auction, max(price)] }
            └── StreamHashAgg [append_only] { group_key: [auction], aggs: [max(price), count] } { result table: 5, state tables: [], distinct tables: [] }
                └── StreamExchange Hash([0]) from 2

    Fragment 1
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 4 }

    Fragment 2
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 6 }

    Table 0
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2 { columns: [ auction, max(price) ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 3 { columns: [ auction, _degree ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 5 { columns: [ auction, max(price), count ], primary key: [ $0 ASC ], value indices: [ 1, 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 6 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction_id, auction_item_name, current_highest_bid, _row_id, auction ]
    ├── primary key: [ $3 ASC, $4 ASC, $0 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 3

- id: nexmark_q102
  before:
  - create_sources
  sql: |
    -- A self-made query that covers dynamic filter.
    --
    -- Show the auctions whose count of bids is greater than the overall average count of bids
    -- per auction.
    SELECT
        a.id AS auction_id,
        a.item_name AS auction_item_name,
        COUNT(b.auction) AS bid_count
    FROM auction a
    JOIN bid b ON a.id = b.auction
    GROUP BY a.id, a.item_name
    HAVING COUNT(b.auction) >= (
        SELECT COUNT(*) / COUNT(DISTINCT auction) FROM bid
    )
  batch_plan: |-
    BatchNestedLoopJoin { type: Inner, predicate: (count(auction) >= $expr1), output: [id, item_name, count(auction)] }
    ├─BatchExchange { order: [], dist: Single }
    │ └─BatchHashAgg { group_key: [id, item_name], aggs: [count(auction)] }
    │   └─BatchHashJoin { type: Inner, predicate: id = auction, output: [id, item_name, auction] }
    │     ├─BatchExchange { order: [], dist: HashShard(id) }
    │     │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
    │     └─BatchExchange { order: [], dist: HashShard(auction) }
    │       └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
    └─BatchProject { exprs: [(sum0(sum0(count)) / sum0(count(auction))) as $expr1] }
      └─BatchSimpleAgg { aggs: [sum0(sum0(count)), sum0(count(auction))] }
        └─BatchExchange { order: [], dist: Single }
          └─BatchSimpleAgg { aggs: [sum0(count), count(auction)] }
            └─BatchHashAgg { group_key: [auction], aggs: [count] }
              └─BatchExchange { order: [], dist: HashShard(auction) }
                └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction_id, auction_item_name, bid_count], stream_key: [auction_id, auction_item_name], pk_columns: [auction_id, auction_item_name], pk_conflict: NoCheck }
    └─StreamDynamicFilter { predicate: (count(auction) >= $expr1), output: [id, item_name, count(auction)] }
      ├─StreamProject { exprs: [id, item_name, count(auction)] }
      │ └─StreamHashAgg [append_only] { group_key: [id, item_name], aggs: [count(auction), count] }
      │   └─StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: [id, item_name, auction, _row_id, _row_id] }
      │     ├─StreamExchange { dist: HashShard(id) }
      │     │ └─StreamRowIdGen { row_id_index: 10 }
      │     │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
      │     └─StreamExchange { dist: HashShard(auction) }
      │       └─StreamShare { id: 5 }
      │         └─StreamProject { exprs: [auction, _row_id] }
      │           └─StreamRowIdGen { row_id_index: 7 }
      │             └─StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] }
      └─StreamExchange { dist: Broadcast }
        └─StreamProject { exprs: [(sum0(sum0(count)) / sum0(count(auction))) as $expr1] }
          └─StreamSimpleAgg { aggs: [sum0(sum0(count)), sum0(count(auction)), count] }
            └─StreamExchange { dist: Single }
              └─StreamStatelessSimpleAgg { aggs: [sum0(count), count(auction)] }
                └─StreamHashAgg [append_only] { group_key: [auction], aggs: [count] }
                  └─StreamExchange { dist: HashShard(auction) }
                    └─StreamShare { id: 5 }
                      └─StreamProject { exprs: [auction, _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_id, auction_item_name, bid_count], stream_key: [auction_id, auction_item_name], pk_columns: [auction_id, auction_item_name], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamDynamicFilter { predicate: (count(auction) >= $expr1), output: [id, item_name, count(auction)] } { left table: 0, right table: 1 }
        ├── StreamProject { exprs: [id, item_name, count(auction)] }
        │   └── StreamHashAgg [append_only] { group_key: [id, item_name], aggs: [count(auction), count] } { result table: 2, state tables: [], distinct tables: [] }
        │       └── StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: [id, item_name, auction, _row_id, _row_id] }
        │           ├── left table: 3
        │           ├── right table: 5
        │           ├── left degree table: 4
        │           ├── right degree table: 6
        │           ├── StreamExchange Hash([0]) from 1
        │           └── StreamExchange Hash([0]) from 2
        └── StreamExchange Broadcast from 4

    Fragment 1
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 7 }

    Fragment 2
    StreamNoOp
    └── StreamExchange NoShuffle from 3

    Fragment 3
    StreamProject { exprs: [auction, _row_id] }
    └── StreamRowIdGen { row_id_index: 7 }
        └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 8 }

    Fragment 4
    StreamProject { exprs: [(sum0(sum0(count)) / sum0(count(auction))) as $expr1] }
    └── StreamSimpleAgg { aggs: [sum0(sum0(count)), sum0(count(auction)), count] } { result table: 9, state tables: [], distinct tables: [] }
        └── StreamExchange Single from 5

    Fragment 5
    StreamStatelessSimpleAgg { aggs: [sum0(count), count(auction)] }
    └── StreamHashAgg [append_only] { group_key: [auction], aggs: [count] } { result table: 10, state tables: [], distinct tables: [] }
        └── StreamExchange Hash([0]) from 6

    Fragment 6
    StreamNoOp
    └── StreamExchange NoShuffle from 3

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

    Table 1 { columns: [ $expr1 ], primary key: [], value indices: [ 0 ], distribution key: [], read pk prefix len hint: 0 }

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

    Table 3
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 4 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 5 { columns: [ auction, _row_id ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 0, 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 6 { columns: [ auction, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 7 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 8 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 9 { columns: [ sum0(sum0(count)), sum0(count(auction)), count ], primary key: [], value indices: [ 0, 1, 2 ], distribution key: [], read pk prefix len hint: 0 }

    Table 10 { columns: [ auction, count ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction_id, auction_item_name, bid_count ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 0, 1, 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 2

- id: nexmark_q103
  before:
  - create_sources
  sql: |
    -- A self-made query that covers semi join.
    --
    -- Show the auctions that have at least 20 bids.
    SELECT
        a.id AS auction_id,
        a.item_name AS auction_item_name
    FROM auction a
    WHERE a.id IN (
        SELECT b.auction FROM bid b
        GROUP BY b.auction
        HAVING COUNT(*) >= 20
    );
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: LeftSemi, predicate: id = auction, output: [id, item_name] }
      ├─BatchExchange { order: [], dist: HashShard(id) }
      │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
      └─BatchProject { exprs: [auction] }
        └─BatchFilter { predicate: (count >= 20:Int32) }
          └─BatchHashAgg { group_key: [auction], aggs: [count] }
            └─BatchExchange { order: [], dist: HashShard(auction) }
              └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction_id, auction_item_name, _row_id(hidden)], stream_key: [_row_id, auction_id], pk_columns: [_row_id, auction_id], pk_conflict: NoCheck }
    └─StreamHashJoin { type: LeftSemi, predicate: id = auction, output: [id, item_name, _row_id] }
      ├─StreamExchange { dist: HashShard(id) }
      │ └─StreamRowIdGen { row_id_index: 10 }
      │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
      └─StreamProject { exprs: [auction] }
        └─StreamFilter { predicate: (count >= 20:Int32) }
          └─StreamHashAgg [append_only] { group_key: [auction], aggs: [count] }
            └─StreamExchange { dist: HashShard(auction) }
              └─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_id, auction_item_name, _row_id(hidden)], stream_key: [_row_id, auction_id], pk_columns: [_row_id, auction_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamHashJoin { type: LeftSemi, predicate: id = auction, output: [id, item_name, _row_id] }
        ├── left table: 0
        ├── right table: 2
        ├── left degree table: 1
        ├── right degree table: 3
        ├── StreamExchange Hash([0]) from 1
        └── StreamProject { exprs: [auction] }
            └── StreamFilter { predicate: (count >= 20:Int32) }
                └── StreamHashAgg [append_only] { group_key: [auction], aggs: [count] } { result table: 5, state tables: [], distinct tables: [] }
                    └── StreamExchange Hash([0]) from 2

    Fragment 1
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
        └── source state table: 4

    Fragment 2
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 6 }

    Table 0
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2 { columns: [ auction ], primary key: [ $0 ASC ], value indices: [ 0 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 3 { columns: [ auction, _degree ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 5 { columns: [ auction, count ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 6 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction_id, auction_item_name, _row_id ]
    ├── primary key: [ $2 ASC, $0 ASC ]
    ├── value indices: [ 0, 1, 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 2

- id: nexmark_q104
  before:
  - create_sources
  sql: |
    -- A self-made query that covers anti join.
    --
    -- This is the same as q103, which shows the auctions that have at least 20 bids.
    SELECT
        a.id AS auction_id,
        a.item_name AS auction_item_name
    FROM auction a
    WHERE a.id NOT IN (
        SELECT b.auction FROM bid b
        GROUP BY b.auction
        HAVING COUNT(*) < 20
    );
  batch_plan: |-
    BatchExchange { order: [], dist: Single }
    └─BatchHashJoin { type: LeftAnti, predicate: id = auction, output: [id, item_name] }
      ├─BatchExchange { order: [], dist: HashShard(id) }
      │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
      └─BatchProject { exprs: [auction] }
        └─BatchFilter { predicate: (count < 20:Int32) }
          └─BatchHashAgg { group_key: [auction], aggs: [count] }
            └─BatchExchange { order: [], dist: HashShard(auction) }
              └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction_id, auction_item_name, _row_id(hidden)], stream_key: [_row_id, auction_id], pk_columns: [_row_id, auction_id], pk_conflict: NoCheck }
    └─StreamHashJoin { type: LeftAnti, predicate: id = auction, output: [id, item_name, _row_id] }
      ├─StreamExchange { dist: HashShard(id) }
      │ └─StreamRowIdGen { row_id_index: 10 }
      │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
      └─StreamProject { exprs: [auction] }
        └─StreamFilter { predicate: (count < 20:Int32) }
          └─StreamHashAgg [append_only] { group_key: [auction], aggs: [count] }
            └─StreamExchange { dist: HashShard(auction) }
              └─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_id, auction_item_name, _row_id(hidden)], stream_key: [_row_id, auction_id], pk_columns: [_row_id, auction_id], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamHashJoin { type: LeftAnti, predicate: id = auction, output: [id, item_name, _row_id] }
        ├── left table: 0
        ├── right table: 2
        ├── left degree table: 1
        ├── right degree table: 3
        ├── StreamExchange Hash([0]) from 1
        └── StreamProject { exprs: [auction] }
            └── StreamFilter { predicate: (count < 20:Int32) }
                └── StreamHashAgg [append_only] { group_key: [auction], aggs: [count] } { result table: 5, state tables: [], distinct tables: [] }
                    └── StreamExchange Hash([0]) from 2

    Fragment 1
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
        └── source state table: 4

    Fragment 2
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 6 }

    Table 0
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 1 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 2 { columns: [ auction ], primary key: [ $0 ASC ], value indices: [ 0 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 3 { columns: [ auction, _degree ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 4 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 5 { columns: [ auction, count ], primary key: [ $0 ASC ], value indices: [ 1 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 6 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294
    ├── columns: [ auction_id, auction_item_name, _row_id ]
    ├── primary key: [ $2 ASC, $0 ASC ]
    ├── value indices: [ 0, 1, 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 2

- id: nexmark_q105
  before:
  - create_sources
  sql: |
    -- A self-made query that covers singleton top-n (and local-phase group top-n).
    --
    -- Show the top 1000 auctions by the number of bids.
    SELECT
        a.id AS auction_id,
        a.item_name AS auction_item_name,
        COUNT(b.auction) AS bid_count
    FROM auction a
    JOIN bid b ON a.id = b.auction
    GROUP BY a.id, a.item_name
    ORDER BY bid_count DESC
    LIMIT 1000;
  batch_plan: |-
    BatchTopN { order: [count(auction) DESC], limit: 1000, offset: 0 }
    └─BatchExchange { order: [], dist: Single }
      └─BatchTopN { order: [count(auction) DESC], limit: 1000, offset: 0 }
        └─BatchHashAgg { group_key: [id, item_name], aggs: [count(auction)] }
          └─BatchHashJoin { type: Inner, predicate: id = auction, output: [id, item_name, auction] }
            ├─BatchExchange { order: [], dist: HashShard(id) }
            │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
            └─BatchExchange { order: [], dist: HashShard(auction) }
              └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [auction_id, auction_item_name, bid_count], stream_key: [auction_id, auction_item_name], pk_columns: [bid_count, auction_id, auction_item_name], pk_conflict: NoCheck }
    └─StreamProject { exprs: [id, item_name, count(auction)] }
      └─StreamTopN { order: [count(auction) DESC], limit: 1000, offset: 0 }
        └─StreamExchange { dist: Single }
          └─StreamGroupTopN { order: [count(auction) DESC], limit: 1000, offset: 0, group_key: [3] }
            └─StreamProject { exprs: [id, item_name, count(auction), Vnode(id) as $expr1] }
              └─StreamHashAgg [append_only] { group_key: [id, item_name], aggs: [count(auction), count] }
                └─StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: [id, item_name, auction, _row_id, _row_id] }
                  ├─StreamExchange { dist: HashShard(id) }
                  │ └─StreamRowIdGen { row_id_index: 10 }
                  │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
                  └─StreamExchange { dist: HashShard(auction) }
                    └─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_id, auction_item_name, bid_count], stream_key: [auction_id, auction_item_name], pk_columns: [bid_count, auction_id, auction_item_name], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [id, item_name, count(auction)] }
        └── StreamTopN { order: [count(auction) DESC], limit: 1000, offset: 0 } { state table: 0 }
            └── StreamExchange Single from 1

    Fragment 1
    StreamGroupTopN { order: [count(auction) DESC], limit: 1000, offset: 0, group_key: [3] } { state table: 1 }
    └── StreamProject { exprs: [id, item_name, count(auction), Vnode(id) as $expr1] }
        └── StreamHashAgg [append_only] { group_key: [id, item_name], aggs: [count(auction), count] } { result table: 2, state tables: [], distinct tables: [] }
            └── StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: [id, item_name, auction, _row_id, _row_id] }
                ├── left table: 3
                ├── right table: 5
                ├── left degree table: 4
                ├── right degree table: 6
                ├── StreamExchange Hash([0]) from 2
                └── StreamExchange Hash([0]) from 3

    Fragment 2
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] } { source state table: 7 }

    Fragment 3
    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 ], value indices: [ 0, 1, 2, 3 ], distribution key: [], read pk prefix len hint: 0 }

    Table 1
    ├── columns: [ id, item_name, count(auction), $expr1 ]
    ├── primary key: [ $3 ASC, $2 DESC, $0 ASC, $1 ASC ]
    ├── value indices: [ 0, 1, 2, 3 ]
    ├── distribution key: [ 0 ]
    ├── read pk prefix len hint: 1
    └── vnode column idx: 3

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

    Table 3
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 4 { columns: [ id, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

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

    Table 6 { columns: [ auction, _row_id, _degree ], primary key: [ $0 ASC, $1 ASC ], value indices: [ 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 7 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 8 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 4294967294 { columns: [ auction_id, auction_item_name, bid_count ], primary key: [ $2 DESC, $0 ASC, $1 ASC ], value indices: [ 0, 1, 2 ], distribution key: [], read pk prefix len hint: 2 }

- id: nexmark_q106
  before:
  - create_sources
  sql: |
    -- A self-made query that covers two-phase stateful simple aggregation.
    --
    -- Show the minimum final price of all auctions.
    SELECT
        MIN(final) AS min_final
    FROM
        (
            SELECT
                auction.id,
                MAX(price) AS final
            FROM
                auction,
                bid
            WHERE
                bid.auction = auction.id
                AND bid.date_time BETWEEN auction.date_time AND auction.expires
            GROUP BY
                auction.id
        )
  batch_plan: |-
    BatchSimpleAgg { aggs: [min(min(max(price)))] }
    └─BatchExchange { order: [], dist: Single }
      └─BatchSimpleAgg { aggs: [min(max(price))] }
        └─BatchHashAgg { group_key: [id], aggs: [max(price)] }
          └─BatchHashJoin { type: Inner, predicate: id = auction AND (date_time >= date_time) AND (date_time <= expires), output: [id, price] }
            ├─BatchExchange { order: [], dist: HashShard(id) }
            │ └─BatchSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id], filter: (None, None) }
            └─BatchExchange { order: [], dist: HashShard(auction) }
              └─BatchSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id], filter: (None, None) }
  stream_plan: |-
    StreamMaterialize { columns: [min_final], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
    └─StreamProject { exprs: [min(min(max(price)))] }
      └─StreamSimpleAgg { aggs: [min(min(max(price))), count] }
        └─StreamExchange { dist: Single }
          └─StreamHashAgg { group_key: [$expr1], aggs: [min(max(price)), count] }
            └─StreamProject { exprs: [id, max(price), Vnode(id) as $expr1] }
              └─StreamHashAgg [append_only] { group_key: [id], aggs: [max(price), count] }
                └─StreamProject { exprs: [id, price, _row_id, _row_id] }
                  └─StreamFilter { predicate: (date_time >= date_time) AND (date_time <= expires) }
                    └─StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: all }
                      ├─StreamExchange { dist: HashShard(id) }
                      │ └─StreamRowIdGen { row_id_index: 10 }
                      │   └─StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
                      └─StreamExchange { dist: HashShard(auction) }
                        └─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: [min_final], stream_key: [], pk_columns: [], pk_conflict: NoCheck }
    ├── materialized table: 4294967294
    └── StreamProject { exprs: [min(min(max(price)))] }
        └── StreamSimpleAgg { aggs: [min(min(max(price))), count] }
            ├── result table: 1
            ├── state tables: [ 0 ]
            ├── distinct tables: []
            └── StreamExchange Single from 1

    Fragment 1
    StreamHashAgg { group_key: [$expr1], aggs: [min(max(price)), count] }
    ├── result table: 3
    ├── state tables: [ 2 ]
    ├── distinct tables: []
    └── StreamProject { exprs: [id, max(price), Vnode(id) as $expr1] }
        └── StreamHashAgg [append_only] { group_key: [id], aggs: [max(price), count] }
            ├── result table: 4
            ├── state tables: []
            ├── distinct tables: []
            └── StreamProject { exprs: [id, price, _row_id, _row_id] }
                └── StreamFilter { predicate: (date_time >= date_time) AND (date_time <= expires) }
                    └── StreamHashJoin [append_only] { type: Inner, predicate: id = auction, output: all }
                        ├── left table: 5
                        ├── right table: 7
                        ├── left degree table: 6
                        ├── right degree table: 8
                        ├── StreamExchange Hash([0]) from 2
                        └── StreamExchange Hash([0]) from 3

    Fragment 2
    StreamRowIdGen { row_id_index: 10 }
    └── StreamSource { source: auction, columns: [id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id] }
        └── source state table: 9

    Fragment 3
    StreamRowIdGen { row_id_index: 7 }
    └── StreamSource { source: bid, columns: [auction, bidder, price, channel, url, date_time, extra, _row_id] } { source state table: 10 }

    Table 0
    ├── columns: [ min(max(price)), $expr1 ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 0, 1 ]
    ├── distribution key: []
    └── read pk prefix len hint: 0

    Table 1 { columns: [ min(min(max(price))), count ], primary key: [], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 0 }

    Table 2
    ├── columns: [ $expr1, max(price), id ]
    ├── primary key: [ $0 ASC, $1 ASC, $2 ASC ]
    ├── value indices: [ 1, 2 ]
    ├── distribution key: [ 2 ]
    ├── read pk prefix len hint: 1
    └── vnode column idx: 0

    Table 3
    ├── columns: [ $expr1, min(max(price)), count ]
    ├── primary key: [ $0 ASC ]
    ├── value indices: [ 1, 2 ]
    ├── distribution key: []
    ├── read pk prefix len hint: 1
    └── vnode column idx: 0

    Table 4 { columns: [ id, max(price), count ], primary key: [ $0 ASC ], value indices: [ 1, 2 ], distribution key: [ 0 ], read pk prefix len hint: 1 }

    Table 5
    ├── columns: [ id, item_name, description, initial_bid, reserve, date_time, expires, seller, category, extra, _row_id ]
    ├── primary key: [ $0 ASC, $10 ASC ]
    ├── value indices: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 6
    ├── columns: [ id, _row_id, _degree ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

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

    Table 8
    ├── columns: [ auction, _row_id, _degree ]
    ├── primary key: [ $0 ASC, $1 ASC ]
    ├── value indices: [ 2 ]
    ├── distribution key: [ 0 ]
    └── read pk prefix len hint: 1

    Table 9 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

    Table 10 { columns: [ partition_id, offset_info ], primary key: [ $0 ASC ], value indices: [ 0, 1 ], distribution key: [], read pk prefix len hint: 1 }

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