lmatz
commented
7 months ago RW Query:
create sink tpch_q1 as
select
l_returnflag,
l_linestatus,
sum(l_quantity) as sum_qty,
sum(l_extendedprice) as sum_base_price,
sum(l_extendedprice * (1 - l_discount)) as sum_disc_price,
sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge,
round(avg(l_quantity), 4) as avg_qty,
round(avg(l_extendedprice), 4) as avg_price,
round(avg(l_discount), 4) as avg_disc,
count(*) as count_order
from
lineitem
where
l_shipdate <= date '2099-12-01' - interval '71' day
group by
l_returnflag,
l_linestatus
with ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');Query Plan:
StreamSink { type: append-only, columns: [l_returnflag, l_linestatus, sum_qty, sum_base_price, sum_disc_price, sum_charge, avg_qty, avg_price, avg_disc, count_order] }
└─StreamProject { exprs: [$expr5, $expr6, sum($expr1), sum($expr2), sum($expr7), sum($expr8), RoundDigit((sum($expr1) / count($expr1)::Decimal), 4:Int32) as $expr9, RoundDigit((sum($expr2) / count($expr2)::Decimal), 4:Int32) as $expr10, RoundDigit((sum($expr3) / count($expr3)::Decimal), 4:Int32) as $expr11, count] }
└─StreamHashAgg [append_only] { group_key: [$expr5, $expr6], aggs: [sum($expr1), sum($expr2), sum($expr7), sum($expr8), count($expr1), count($expr2), sum($expr3), count($expr3), count] }
└─StreamExchange { dist: HashShard($expr5, $expr6) }
└─StreamProject { exprs: [$expr5, $expr6, $expr1, $expr2, $expr7, ($expr7 * (1:Decimal + $expr4)) as $expr8, $expr3, _row_id] }
└─StreamProject { exprs: [$expr1, $expr2, $expr3, $expr4, $expr5, $expr6, ($expr2 * (1:Decimal - $expr3)) as $expr7, _row_id] }
└─StreamProject { exprs: [Field(lineitem, 4:Int32) as $expr1, Field(lineitem, 5:Int32) as $expr2, Field(lineitem, 6:Int32) as $expr3, Field(lineitem, 7:Int32) as $expr4, Field(lineitem, 8:Int32) as $expr5, Field(lineitem, 9:Int32) as $expr6, _row_id] }
└─StreamFilter { predicate: (Field(lineitem, 10:Int32) <= '2099-09-21 00:00:00':Timestamp) AND (eventType = 'lineitem':Varchar) }
└─StreamRowIdGen { row_id_index: 10 }
└─StreamSource { source: tpch, columns: [eventType, lineitem, supplier, part, partsupp, customer, orders, nation, region, _rw_kafka_timestamp, _row_id] }
(10 rows)Flink:
INSERT INTO tpch_q1
select
l_returnflag,
l_linestatus,
sum(l_quantity) as sum_qty,
sum(l_extendedprice) as sum_base_price,
sum(l_extendedprice * (1 - l_discount)) as sum_disc_price,
sum(l_extendedprice * (1 - l_discount) * (1 + l_tax)) as sum_charge,
round(avg(l_quantity), 4) as avg_qty,
round(avg(l_extendedprice), 4) as avg_price,
round(avg(l_discount), 4) as avg_disc,
count(*) as count_order
from
lineitem
where
l_shipdate <= date '2099-12-01' - interval '71' day
group by
l_returnflag,
l_linestatus;Query Plan:
== Optimized Physical Plan ==
Sink(table=[default_catalog.default_database.tpch_q1], fields=[l_returnflag, l_linestatus, sum_qty, sum_base_price, sum_disc_price, sum_charge, avg_qty, avg_price, avg_disc, count_order])
+- Calc(select=[l_returnflag, l_linestatus, CAST(sum_qty AS DECIMAL(10, 0)) AS sum_qty, CAST(sum_base_price AS DECIMAL(10, 0)) AS sum_base_price, CAST(sum_disc_price AS DECIMAL(10, 0)) AS sum_disc_price, CAST(sum_charge AS DECIMAL(10, 0)) AS sum_charge, CAST(ROUND($f6, 4) AS DECIMAL(10, 0)) AS avg_qty, CAST(ROUND($f7, 4) AS DECIMAL(10, 0)) AS avg_price, CAST(ROUND($f8, 4) AS DECIMAL(10, 0)) AS avg_disc, CAST(count_order AS BIGINT) AS count_order])
+- GroupAggregate(groupBy=[l_returnflag, l_linestatus], select=[l_returnflag, l_linestatus, SUM(l_quantity) AS sum_qty, SUM(l_extendedprice) AS sum_base_price, SUM($f4) AS sum_disc_price, SUM($f5) AS sum_charge, AVG(l_quantity) AS $f6, AVG(l_extendedprice) AS $f7, AVG(l_discount) AS $f8, COUNT(*) AS count_order])
+- Exchange(distribution=[hash[l_returnflag, l_linestatus]])
+- Calc(select=[lineitem.l_returnflag AS l_returnflag, lineitem.l_linestatus AS l_linestatus, lineitem.l_quantity AS l_quantity, lineitem.l_extendedprice AS l_extendedprice, *(lineitem.l_extendedprice, -(1, lineitem.l_discount)) AS $f4, *(*(lineitem.l_extendedprice, -(1, lineitem.l_discount)), +(1, lineitem.l_tax)) AS $f5, lineitem.l_discount AS l_discount], where=[AND(=(eventType, _UTF-16LE'lineitem':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), <=(lineitem.l_shipdate, 2099-09-21))])
+- TableSourceScan(table=[[default_catalog, default_database, tpch]], fields=[eventType, lineitem, supplier, part, partsupp, customer, orders, nation, region])
== Optimized Execution Plan ==
Sink(table=[default_catalog.default_database.tpch_q1], fields=[l_returnflag, l_linestatus, sum_qty, sum_base_price, sum_disc_price, sum_charge, avg_qty, avg_price, avg_disc, count_order])
+- Calc(select=[l_returnflag, l_linestatus, CAST(sum_qty AS DECIMAL(10, 0)) AS sum_qty, CAST(sum_base_price AS DECIMAL(10, 0)) AS sum_base_price, CAST(sum_disc_price AS DECIMAL(10, 0)) AS sum_disc_price, CAST(sum_charge AS DECIMAL(10, 0)) AS sum_charge, CAST(ROUND($f6, 4) AS DECIMAL(10, 0)) AS avg_qty, CAST(ROUND($f7, 4) AS DECIMAL(10, 0)) AS avg_price, CAST(ROUND($f8, 4) AS DECIMAL(10, 0)) AS avg_disc, CAST(count_order AS BIGINT) AS count_order])
+- GroupAggregate(groupBy=[l_returnflag, l_linestatus], select=[l_returnflag, l_linestatus, SUM(l_quantity) AS sum_qty, SUM(l_extendedprice) AS sum_base_price, SUM($f4) AS sum_disc_price, SUM($f5) AS sum_charge, AVG(l_quantity) AS $f6, AVG(l_extendedprice) AS $f7, AVG(l_discount) AS $f8, COUNT(*) AS count_order])
+- Exchange(distribution=[hash[l_returnflag, l_linestatus]])
+- Calc(select=[lineitem.l_returnflag AS l_returnflag, lineitem.l_linestatus AS l_linestatus, lineitem.l_quantity AS l_quantity, lineitem.l_extendedprice AS l_extendedprice, (lineitem.l_extendedprice * (1 - lineitem.l_discount)) AS $f4, ((lineitem.l_extendedprice * (1 - lineitem.l_discount)) * (1 + lineitem.l_tax)) AS $f5, lineitem.l_discount AS l_discount], where=[((eventType = 'lineitem') AND (lineitem.l_shipdate <= 2099-09-21))])
+- TableSourceScan(table=[[default_catalog, default_database, tpch]], fields=[eventType, lineitem, supplier, part, partsupp, customer, orders, nation, region])
st1page
github-actions[bot]
See performance numbers at https://www.notion.so/risingwave-labs/TPCH-Performance-Numbers-Table-e098ef82884546949333409f0513ada7?pvs=4#8de0bf4bda51444c8381f3b0c10ddfe1