You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/16 04:17:18 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-7044-a45445c6ef770b2d008b63ef42478bfdbc3f1e01 created (now c230bf4e98)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7044-a45445c6ef770b2d008b63ef42478bfdbc3f1e01
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at c230bf4e98 feat: add nested benchmark derived from TPC-H (#7044)
This branch includes the following new commits:
new c230bf4e98 feat: add nested benchmark derived from TPC-H (#7044)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: feat: add nested benchmark derived from TPC-H (#7044)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7044-a45445c6ef770b2d008b63ef42478bfdbc3f1e01
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit c230bf4e986ecfaf317c0621b13cd84931c948ae
Author: Oleks V <co...@users.noreply.github.com>
AuthorDate: Fri Oct 16 04:16:55 2026 +0000
feat: add nested benchmark derived from TPC-H (#7044)
Adds benchmarks/nested: a spark-shell script that converts existing flat
TPC-H data (Parquet or Iceberg) into a nested layout, where non-key,
non-partition columns move into a single-element array<struct>, and the 22
queries rewritten for that schema.
---
benchmarks/nested/README.md | 195 +++++++++++++++++++++
benchmarks/nested/queries/q01.sql | 25 +++
benchmarks/nested/queries/q02.sql | 48 +++++
benchmarks/nested/queries/q03.sql | 27 +++
benchmarks/nested/queries/q04.sql | 25 +++
benchmarks/nested/queries/q05.sql | 28 +++
benchmarks/nested/queries/q06.sql | 13 ++
benchmarks/nested/queries/q07.sql | 43 +++++
benchmarks/nested/queries/q08.sql | 41 +++++
benchmarks/nested/queries/q09.sql | 36 ++++
benchmarks/nested/queries/q10.sql | 36 ++++
benchmarks/nested/queries/q11.sql | 31 ++++
benchmarks/nested/queries/q12.sql | 32 ++++
benchmarks/nested/queries/q13.sql | 24 +++
benchmarks/nested/queries/q14.sql | 17 ++
benchmarks/nested/queries/q15.sql | 37 ++++
benchmarks/nested/queries/q16.sql | 34 ++++
benchmarks/nested/queries/q17.sql | 21 +++
benchmarks/nested/queries/q18.sql | 37 ++++
benchmarks/nested/queries/q19.sql | 39 +++++
benchmarks/nested/queries/q20.sql | 41 +++++
benchmarks/nested/queries/q21.sql | 44 +++++
benchmarks/nested/queries/q22.sql | 41 +++++
benchmarks/nested/scripts/nest_tpch.scala | 279 ++++++++++++++++++++++++++++++
dev/release/rat_exclude_files.txt | 1 +
25 files changed, 1195 insertions(+)
diff --git a/benchmarks/nested/README.md b/benchmarks/nested/README.md
new file mode 100644
index 0000000000..aef58a6753
--- /dev/null
+++ b/benchmarks/nested/README.md
@@ -0,0 +1,195 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Nested Benchmark Derived from TPC-H
+
+TPC-H data and queries with a nested schema, for benchmarking how engines handle nested data.
+The queries are derived from TPC-H under the TPC Fair Use Policy.
+
+## Motivation
+
+TPC-H is flat. All eight tables are normalized and every column is primitive. Real-world
+data usually isn't: event logs, JSON/Avro/Protobuf records and data lake tables are full of
+structs and arrays. Nested column pruning, struct predicate pushdown and array unnesting
+are hot paths in practice, but flat TPC-H exercises none of them.
+
+Nested TPC-H closes that gap. It keeps the TPC-H data and queries and changes only the
+shape. Each array holds exactly one element, so the nested queries return the same answers
+as the flat ones.
+
+```mermaid
+flowchart LR
+ gen["TPC-H generator"] --> flat[("Flat TPC-H<br/>Parquet / Iceberg")]
+ flat --> nest["scripts/nest_tpch.scala<br/>(Spark)"]
+ nest --> nested[("Nested TPC-H<br/>Parquet / Iceberg")]
+ nested --> queries["queries/q01..q22.sql"]
+ queries --> engines["Spark, Comet, Gluten, Trino, ..."]
+```
+
+## Schema
+
+Join keys and partition columns stay at the top level. All other columns move into
+`<table>_data: array<struct<...>>`.
+
+```mermaid
+flowchart LR
+ subgraph flat["lineitem (flat)"]
+ direction TB
+ f1["l_orderkey, l_partkey, l_suppkey"]
+ f2["l_shipdate"]
+ f3["l_linenumber, l_quantity, l_extendedprice,<br/>l_discount, l_tax, l_returnflag, l_linestatus,<br/>l_commitdate, l_receiptdate, l_shipinstruct,<br/>l_shipmode, l_comment"]
+ end
+ subgraph nested["lineitem (nested)"]
+ direction TB
+ n1["l_orderkey, l_partkey, l_suppkey"]
+ n2["l_shipdate"]
+ n3["lineitem_data: ARRAY#60;STRUCT#60;<br/>l_linenumber, l_quantity, l_extendedprice,<br/>l_discount, l_tax, l_returnflag, l_linestatus,<br/>l_commitdate, l_receiptdate, l_shipinstruct,<br/>l_shipmode, l_comment<br/>#62;#62;"]
+ end
+ f1 -- "join keys" --> n1
+ f2 -- "partition column" --> n2
+ f3 -- "everything else" --> n3
+```
+
+For example, `lineitem`. The keys `l_orderkey`, `l_partkey` and `l_suppkey` and the
+partition column `l_shipdate` stay at the top level. The rest moves into `lineitem_data`.
+
+Flat:
+
+```
+l_orderkey BIGINT
+l_partkey BIGINT
+l_suppkey BIGINT
+l_linenumber INT
+l_quantity DECIMAL(12,2)
+l_extendedprice DECIMAL(12,2)
+l_discount DECIMAL(12,2)
+l_tax DECIMAL(12,2)
+l_returnflag STRING
+l_linestatus STRING
+l_shipdate DATE
+l_commitdate DATE
+l_receiptdate DATE
+l_shipinstruct STRING
+l_shipmode STRING
+l_comment STRING
+```
+
+Nested:
+
+```
+l_orderkey BIGINT
+l_partkey BIGINT
+l_suppkey BIGINT
+l_shipdate DATE
+lineitem_data ARRAY<STRUCT<
+ l_linenumber: INT,
+ l_quantity: DECIMAL(12,2),
+ l_extendedprice: DECIMAL(12,2),
+ l_discount: DECIMAL(12,2),
+ l_tax: DECIMAL(12,2),
+ l_returnflag: STRING,
+ l_linestatus: STRING,
+ l_commitdate: DATE,
+ l_receiptdate: DATE,
+ l_shipinstruct: STRING,
+ l_shipmode: STRING,
+ l_comment: STRING>>
+```
+
+The other tables follow the same rule. The partition columns are `p_brand` (`part`),
+`c_mktsegment` (`customer`), `o_orderdate` (`orders`) and `l_shipdate` (`lineitem`).
+
+Source columns are cast to fixed types: `BIGINT` keys, `DECIMAL(12,2)` money, `DATE`,
+`STRING`. Output is written with zstd compression and 512 MiB row groups. Run the script
+with `dryRun=true` to print the full DDL.
+
+## Prerequisites
+
+- An already generated flat TPC-H dataset in Parquet or Iceberg. If you don't have one,
+ generate it with [tpcgen-cli](https://github.com/datafusion-contrib/tpcgen-rs)
+ (`tpcgen-cli tpch parquet -s 100 --output-dir /data/tpch/sf100`) or
+ [tpch-dbgen](https://github.com/databricks/tpch-dbgen). Parquet tables are read from
+ `<path>/<table>/` or `<path>/<table>.parquet`.
+- Apache Spark 3.5+ (tested with 3.5.5 and 4.1.1), local or on a cluster sized for your
+ scale factor. For Iceberg, add the matching `iceberg-spark-runtime` package.
+
+## Usage
+
+TPC-H goes up to SF10000, which is about 10 TB and 60 billion `lineitem` rows. Converting
+data at that size can be expensive, so it needs a distributed engine that reads and writes
+Parquet and Iceberg natively on any storage. This benchmark uses Apache Spark for the conversion. The same
+script runs on a laptop or a cluster.
+
+Run the commands below from `benchmarks/nested`. The script needs no build step.
+
+Parquet to Parquet:
+
+```bash
+spark-shell -i scripts/nest_tpch.scala \
+ --conf spark.nesttpch.source=parquet:/data/tpch/sf10 \
+ --conf spark.nesttpch.target=parquet:/data/tpch-nested/sf10
+```
+
+Parquet to Iceberg. Any Iceberg catalog works:
+
+```bash
+spark-shell \
+ --packages org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1 \
+ --conf spark.sql.catalog.nested=org.apache.iceberg.spark.SparkCatalog \
+ --conf spark.sql.catalog.nested.type=hadoop \
+ --conf spark.sql.catalog.nested.warehouse=s3a://bucket/warehouse \
+ -i scripts/nest_tpch.scala \
+ --conf spark.nesttpch.source=parquet:s3a://bucket/tpch/sf1000 \
+ --conf spark.nesttpch.target=iceberg:nested.tpch
+```
+
+For an Iceberg source, use `spark.nesttpch.source=iceberg:<catalog>.<db>`.
+
+| Setting | Default | Description |
+| ---------------------------- | ---------- | ----------------------------------------------- |
+| `spark.nesttpch.source` | (required) | `parquet:<path>` or `iceberg:<catalog>.<db>` |
+| `spark.nesttpch.target` | (required) | Same forms |
+| `spark.nesttpch.tables` | all | Comma separated subset, e.g. `region,nation` |
+| `spark.nesttpch.partitioned` | `true` | Partition by the partition columns listed above |
+| `spark.nesttpch.overwrite` | `true` | Replace existing target tables |
+| `spark.nesttpch.verify` | `true` | Check source and target row counts |
+| `spark.nesttpch.dryRun` | `false` | Print the DDL only |
+
+For SF1000 and above, set `spark.sql.shuffle.partitions` to a few thousand.
+
+## Queries
+
+The queries refer to tables by their bare TPC-H names. Each query unnests the arrays it
+needs:
+
+```sql
+from (select l_orderkey, l_partkey, l_suppkey, l_shipdate,
+ explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+where lineitem_data.l_quantity < 24
+```
+
+To run them in `spark-shell`:
+
+```scala
+Seq("region", "nation", "supplier", "part", "partsupp", "customer", "orders", "lineitem")
+ .foreach(t => spark.read.parquet(s"/data/tpch-nested/sf10/$t").createOrReplaceTempView(t))
+spark.sql(scala.io.Source.fromFile("queries/q01.sql").mkString).show()
+```
+
+TPC-H is a trademark of the TPC. Results from these queries are not TPC-H results.
diff --git a/benchmarks/nested/queries/q01.sql b/benchmarks/nested/queries/q01.sql
new file mode 100644
index 0000000000..3d2e2c44e0
--- /dev/null
+++ b/benchmarks/nested/queries/q01.sql
@@ -0,0 +1,25 @@
+-- CometBench-H nested query 1 derived from TPC-H query 1 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ lineitem_data.l_returnflag,
+ lineitem_data.l_linestatus,
+ sum(lineitem_data.l_quantity) as sum_qty,
+ sum(lineitem_data.l_extendedprice) as sum_base_price,
+ sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)) as sum_disc_price,
+ sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount) * (1 + lineitem_data.l_tax)) as sum_charge,
+ avg(lineitem_data.l_quantity) as avg_qty,
+ avg(lineitem_data.l_extendedprice) as avg_price,
+ avg(lineitem_data.l_discount) as avg_disc,
+ count(*) as count_order
+from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+where
+ l_shipdate <= date '1998-12-01' - interval '90' day
+group by
+ lineitem_data.l_returnflag,
+ lineitem_data.l_linestatus
+order by
+ lineitem_data.l_returnflag,
+ lineitem_data.l_linestatus
diff --git a/benchmarks/nested/queries/q02.sql b/benchmarks/nested/queries/q02.sql
new file mode 100644
index 0000000000..247a78d39f
--- /dev/null
+++ b/benchmarks/nested/queries/q02.sql
@@ -0,0 +1,48 @@
+-- CometBench-H nested query 2 derived from TPC-H query 2 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ supplier_data.s_acctbal,
+ supplier_data.s_name,
+ nation_data.n_name,
+ p_partkey,
+ part_data.p_mfgr,
+ supplier_data.s_address,
+ supplier_data.s_phone,
+ supplier_data.s_comment
+from
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part,
+ (select s_suppkey, s_nationkey, explode(supplier_data) as supplier_data from supplier) as supplier,
+ (select ps_partkey, ps_suppkey, explode(partsupp_data) as partsupp_data from partsupp) as partsupp,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation,
+ (select r_regionkey, explode(region_data) as region_data from region) as region
+where
+ p_partkey = ps_partkey
+ and s_suppkey = ps_suppkey
+ and part_data.p_size = 15
+ and part_data.p_type like '%BRASS'
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and region_data.r_name = 'EUROPE'
+ and partsupp_data.ps_supplycost = (
+ select
+ min(partsupp_data.ps_supplycost)
+ from
+ (select ps_partkey, ps_suppkey, explode(partsupp_data) as partsupp_data from partsupp) as partsupp,
+ supplier,
+ nation,
+ (select r_regionkey, explode(region_data) as region_data from region) as region
+ where
+ p_partkey = ps_partkey
+ and s_suppkey = ps_suppkey
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and region_data.r_name = 'EUROPE'
+ )
+order by
+ supplier_data.s_acctbal desc,
+ nation_data.n_name,
+ supplier_data.s_name,
+ p_partkey
+limit 100
diff --git a/benchmarks/nested/queries/q03.sql b/benchmarks/nested/queries/q03.sql
new file mode 100644
index 0000000000..6b6c078113
--- /dev/null
+++ b/benchmarks/nested/queries/q03.sql
@@ -0,0 +1,27 @@
+-- CometBench-H nested query 3 derived from TPC-H query 3 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ l_orderkey,
+ sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)) as revenue,
+ o_orderdate,
+ orders_data.o_shippriority
+from
+ customer,
+ (select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+where
+ c_mktsegment = 'BUILDING'
+ and c_custkey = o_custkey
+ and l_orderkey = o_orderkey
+ and o_orderdate < date '1995-03-15'
+ and l_shipdate > date '1995-03-15'
+group by
+ l_orderkey,
+ o_orderdate,
+ orders_data.o_shippriority
+order by
+ revenue desc,
+ o_orderdate
+limit 10
diff --git a/benchmarks/nested/queries/q04.sql b/benchmarks/nested/queries/q04.sql
new file mode 100644
index 0000000000..09e6d81e05
--- /dev/null
+++ b/benchmarks/nested/queries/q04.sql
@@ -0,0 +1,25 @@
+-- CometBench-H nested query 4 derived from TPC-H query 4 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ orders_data.o_orderpriority,
+ count(*) as order_count
+from
+ (select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders
+where
+ o_orderdate >= date '1993-07-01'
+ and o_orderdate < date '1993-07-01' + interval '3' month
+ and exists (
+ select
+ *
+ from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+ where
+ l_orderkey = o_orderkey
+ and lineitem_data.l_commitdate < lineitem_data.l_receiptdate
+ )
+group by
+ orders_data.o_orderpriority
+order by
+ orders_data.o_orderpriority
diff --git a/benchmarks/nested/queries/q05.sql b/benchmarks/nested/queries/q05.sql
new file mode 100644
index 0000000000..e0e7995286
--- /dev/null
+++ b/benchmarks/nested/queries/q05.sql
@@ -0,0 +1,28 @@
+-- CometBench-H nested query 5 derived from TPC-H query 5 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ nation_data.n_name,
+ sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)) as revenue
+from
+ customer,
+ orders,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ supplier,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation,
+ (select r_regionkey, explode(region_data) as region_data from region) as region
+where
+ c_custkey = o_custkey
+ and l_orderkey = o_orderkey
+ and l_suppkey = s_suppkey
+ and c_nationkey = s_nationkey
+ and s_nationkey = n_nationkey
+ and n_regionkey = r_regionkey
+ and region_data.r_name = 'ASIA'
+ and o_orderdate >= date '1994-01-01'
+ and o_orderdate < date '1994-01-01' + interval '1' year
+group by
+ nation_data.n_name
+order by
+ revenue desc
diff --git a/benchmarks/nested/queries/q06.sql b/benchmarks/nested/queries/q06.sql
new file mode 100644
index 0000000000..c45c6a2eb9
--- /dev/null
+++ b/benchmarks/nested/queries/q06.sql
@@ -0,0 +1,13 @@
+-- CometBench-H nested query 6 derived from TPC-H query 6 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ sum(lineitem_data.l_extendedprice * lineitem_data.l_discount) as revenue
+from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+where
+ l_shipdate >= date '1994-01-01'
+ and l_shipdate < date '1994-01-01' + interval '1' year
+ and lineitem_data.l_discount between .06 - 0.01 and .06 + 0.01
+ and lineitem_data.l_quantity < 24
diff --git a/benchmarks/nested/queries/q07.sql b/benchmarks/nested/queries/q07.sql
new file mode 100644
index 0000000000..c1e97b3ef5
--- /dev/null
+++ b/benchmarks/nested/queries/q07.sql
@@ -0,0 +1,43 @@
+-- CometBench-H nested query 7 derived from TPC-H query 7 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ supp_nation,
+ cust_nation,
+ l_year,
+ sum(volume) as revenue
+from
+ (
+ select
+ n1.nation_data.n_name as supp_nation,
+ n2.nation_data.n_name as cust_nation,
+ year(l_shipdate) as l_year,
+ lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount) as volume
+ from
+ supplier,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ orders,
+ customer,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as n1,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as n2
+ where
+ s_suppkey = l_suppkey
+ and o_orderkey = l_orderkey
+ and c_custkey = o_custkey
+ and s_nationkey = n1.n_nationkey
+ and c_nationkey = n2.n_nationkey
+ and (
+ (n1.nation_data.n_name = 'FRANCE' and n2.nation_data.n_name = 'GERMANY')
+ or (n1.nation_data.n_name = 'GERMANY' and n2.nation_data.n_name = 'FRANCE')
+ )
+ and l_shipdate between date '1995-01-01' and date '1996-12-31'
+ ) as shipping
+group by
+ supp_nation,
+ cust_nation,
+ l_year
+order by
+ supp_nation,
+ cust_nation,
+ l_year
diff --git a/benchmarks/nested/queries/q08.sql b/benchmarks/nested/queries/q08.sql
new file mode 100644
index 0000000000..af225c2a2a
--- /dev/null
+++ b/benchmarks/nested/queries/q08.sql
@@ -0,0 +1,41 @@
+-- CometBench-H nested query 8 derived from TPC-H query 8 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ o_year,
+ sum(case
+ when nation = 'BRAZIL' then volume
+ else 0
+ end) / sum(volume) as mkt_share
+from
+ (
+ select
+ year(o_orderdate) as o_year,
+ lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount) as volume,
+ n2.nation_data.n_name as nation
+ from
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part,
+ supplier,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ orders,
+ customer,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as n1,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as n2,
+ (select r_regionkey, explode(region_data) as region_data from region) as region
+ where
+ p_partkey = l_partkey
+ and s_suppkey = l_suppkey
+ and l_orderkey = o_orderkey
+ and o_custkey = c_custkey
+ and c_nationkey = n1.n_nationkey
+ and n1.n_regionkey = r_regionkey
+ and region_data.r_name = 'AMERICA'
+ and s_nationkey = n2.n_nationkey
+ and o_orderdate between date '1995-01-01' and date '1996-12-31'
+ and part_data.p_type = 'ECONOMY ANODIZED STEEL'
+ ) as all_nations
+group by
+ o_year
+order by
+ o_year
diff --git a/benchmarks/nested/queries/q09.sql b/benchmarks/nested/queries/q09.sql
new file mode 100644
index 0000000000..4482b145e7
--- /dev/null
+++ b/benchmarks/nested/queries/q09.sql
@@ -0,0 +1,36 @@
+-- CometBench-H nested query 9 derived from TPC-H query 9 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ nation,
+ o_year,
+ sum(amount) as sum_profit
+from
+ (
+ select
+ nation_data.n_name as nation,
+ year(o_orderdate) as o_year,
+ lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount) - partsupp_data.ps_supplycost * lineitem_data.l_quantity as amount
+ from
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part,
+ supplier,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ (select ps_partkey, ps_suppkey, explode(partsupp_data) as partsupp_data from partsupp) as partsupp,
+ orders,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
+ where
+ s_suppkey = l_suppkey
+ and ps_suppkey = l_suppkey
+ and ps_partkey = l_partkey
+ and p_partkey = l_partkey
+ and o_orderkey = l_orderkey
+ and s_nationkey = n_nationkey
+ and part_data.p_name like '%green%'
+ ) as profit
+group by
+ nation,
+ o_year
+order by
+ nation,
+ o_year desc
diff --git a/benchmarks/nested/queries/q10.sql b/benchmarks/nested/queries/q10.sql
new file mode 100644
index 0000000000..784e3e7195
--- /dev/null
+++ b/benchmarks/nested/queries/q10.sql
@@ -0,0 +1,36 @@
+-- CometBench-H nested query 10 derived from TPC-H query 10 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ c_custkey,
+ customer_data.c_name,
+ sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)) as revenue,
+ customer_data.c_acctbal,
+ nation_data.n_name,
+ customer_data.c_address,
+ customer_data.c_phone,
+ customer_data.c_comment
+from
+ (select c_custkey, c_nationkey, c_mktsegment, explode(customer_data) as customer_data from customer) as customer,
+ orders,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
+where
+ c_custkey = o_custkey
+ and l_orderkey = o_orderkey
+ and o_orderdate >= date '1993-10-01'
+ and o_orderdate < date '1993-10-01' + interval '3' month
+ and lineitem_data.l_returnflag = 'R'
+ and c_nationkey = n_nationkey
+group by
+ c_custkey,
+ customer_data.c_name,
+ customer_data.c_acctbal,
+ customer_data.c_phone,
+ nation_data.n_name,
+ customer_data.c_address,
+ customer_data.c_comment
+order by
+ revenue desc
+limit 20
diff --git a/benchmarks/nested/queries/q11.sql b/benchmarks/nested/queries/q11.sql
new file mode 100644
index 0000000000..f3d8d1f3ab
--- /dev/null
+++ b/benchmarks/nested/queries/q11.sql
@@ -0,0 +1,31 @@
+-- CometBench-H nested query 11 derived from TPC-H query 11 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ ps_partkey,
+ sum(partsupp_data.ps_supplycost * partsupp_data.ps_availqty) as value
+from
+ (select ps_partkey, ps_suppkey, explode(partsupp_data) as partsupp_data from partsupp) as partsupp,
+ supplier,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
+where
+ ps_suppkey = s_suppkey
+ and s_nationkey = n_nationkey
+ and nation_data.n_name = 'GERMANY'
+group by
+ ps_partkey having
+ sum(partsupp_data.ps_supplycost * partsupp_data.ps_availqty) > (
+ select
+ sum(partsupp_data.ps_supplycost * partsupp_data.ps_availqty) * 0.0001000000
+ from
+ (select ps_partkey, ps_suppkey, explode(partsupp_data) as partsupp_data from partsupp) as partsupp,
+ supplier,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
+ where
+ ps_suppkey = s_suppkey
+ and s_nationkey = n_nationkey
+ and nation_data.n_name = 'GERMANY'
+ )
+order by
+ value desc
diff --git a/benchmarks/nested/queries/q12.sql b/benchmarks/nested/queries/q12.sql
new file mode 100644
index 0000000000..165a57d9e4
--- /dev/null
+++ b/benchmarks/nested/queries/q12.sql
@@ -0,0 +1,32 @@
+-- CometBench-H nested query 12 derived from TPC-H query 12 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ lineitem_data.l_shipmode,
+ sum(case
+ when orders_data.o_orderpriority = '1-URGENT'
+ or orders_data.o_orderpriority = '2-HIGH'
+ then 1
+ else 0
+ end) as high_line_count,
+ sum(case
+ when orders_data.o_orderpriority <> '1-URGENT'
+ and orders_data.o_orderpriority <> '2-HIGH'
+ then 1
+ else 0
+ end) as low_line_count
+from
+ (select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+where
+ o_orderkey = l_orderkey
+ and lineitem_data.l_shipmode in ('MAIL', 'SHIP')
+ and lineitem_data.l_commitdate < lineitem_data.l_receiptdate
+ and l_shipdate < lineitem_data.l_commitdate
+ and lineitem_data.l_receiptdate >= date '1994-01-01'
+ and lineitem_data.l_receiptdate < date '1994-01-01' + interval '1' year
+group by
+ lineitem_data.l_shipmode
+order by
+ lineitem_data.l_shipmode
diff --git a/benchmarks/nested/queries/q13.sql b/benchmarks/nested/queries/q13.sql
new file mode 100644
index 0000000000..dc8bf5db80
--- /dev/null
+++ b/benchmarks/nested/queries/q13.sql
@@ -0,0 +1,24 @@
+-- CometBench-H nested query 13 derived from TPC-H query 13 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ c_count,
+ count(*) as custdist
+from
+ (
+ select
+ c_custkey,
+ count(o_orderkey) as c_count
+ from
+ customer left outer join (select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders on
+ c_custkey = o_custkey
+ and orders_data.o_comment not like '%special%requests%'
+ group by
+ c_custkey
+ ) as c_orders
+group by
+ c_count
+order by
+ custdist desc,
+ c_count desc
diff --git a/benchmarks/nested/queries/q14.sql b/benchmarks/nested/queries/q14.sql
new file mode 100644
index 0000000000..4d3e47899c
--- /dev/null
+++ b/benchmarks/nested/queries/q14.sql
@@ -0,0 +1,17 @@
+-- CometBench-H nested query 14 derived from TPC-H query 14 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ 100.00 * sum(case
+ when part_data.p_type like 'PROMO%'
+ then lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)
+ else 0
+ end) / sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)) as promo_revenue
+from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part
+where
+ l_partkey = p_partkey
+ and l_shipdate >= date '1995-09-01'
+ and l_shipdate < date '1995-09-01' + interval '1' month
diff --git a/benchmarks/nested/queries/q15.sql b/benchmarks/nested/queries/q15.sql
new file mode 100644
index 0000000000..646acf4455
--- /dev/null
+++ b/benchmarks/nested/queries/q15.sql
@@ -0,0 +1,37 @@
+-- CometBench-H nested query 15 derived from TPC-H query 15 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+with revenue0 as
+ (select
+ l_suppkey as supplier_no,
+ sum(lineitem_data.l_extendedprice * (1 - lineitem_data.l_discount)) as total_revenue
+ from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+ where
+ l_shipdate >= date '1996-01-01'
+ and l_shipdate < date '1996-01-01' + interval '3' month
+ group by
+ l_suppkey)
+
+
+select
+ s_suppkey,
+ supplier_data.s_name,
+ supplier_data.s_address,
+ supplier_data.s_phone,
+ total_revenue
+from
+ (select s_suppkey, s_nationkey, explode(supplier_data) as supplier_data from supplier) as supplier,
+ revenue0
+where
+ s_suppkey = supplier_no
+ and total_revenue = (
+ select
+ max(total_revenue)
+ from
+ revenue0
+ )
+order by
+ s_suppkey
+
diff --git a/benchmarks/nested/queries/q16.sql b/benchmarks/nested/queries/q16.sql
new file mode 100644
index 0000000000..cf4288ff96
--- /dev/null
+++ b/benchmarks/nested/queries/q16.sql
@@ -0,0 +1,34 @@
+-- CometBench-H nested query 16 derived from TPC-H query 16 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ p_brand,
+ part_data.p_type,
+ part_data.p_size,
+ count(distinct ps_suppkey) as supplier_cnt
+from
+ partsupp,
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part
+where
+ p_partkey = ps_partkey
+ and p_brand <> 'Brand#45'
+ and part_data.p_type not like 'MEDIUM POLISHED%'
+ and part_data.p_size in (49, 14, 23, 45, 19, 3, 36, 9)
+ and ps_suppkey not in (
+ select
+ s_suppkey
+ from
+ (select s_suppkey, s_nationkey, explode(supplier_data) as supplier_data from supplier) as supplier
+ where
+ supplier_data.s_comment like '%Customer%Complaints%'
+ )
+group by
+ p_brand,
+ part_data.p_type,
+ part_data.p_size
+order by
+ supplier_cnt desc,
+ p_brand,
+ part_data.p_type,
+ part_data.p_size
diff --git a/benchmarks/nested/queries/q17.sql b/benchmarks/nested/queries/q17.sql
new file mode 100644
index 0000000000..6ba6e997d5
--- /dev/null
+++ b/benchmarks/nested/queries/q17.sql
@@ -0,0 +1,21 @@
+-- CometBench-H nested query 17 derived from TPC-H query 17 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ sum(lineitem_data.l_extendedprice) / 7.0 as avg_yearly
+from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part
+where
+ p_partkey = l_partkey
+ and p_brand = 'Brand#23'
+ and part_data.p_container = 'MED BOX'
+ and lineitem_data.l_quantity < (
+ select
+ 0.2 * avg(lineitem_data.l_quantity)
+ from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+ where
+ l_partkey = p_partkey
+ )
diff --git a/benchmarks/nested/queries/q18.sql b/benchmarks/nested/queries/q18.sql
new file mode 100644
index 0000000000..7eaa89abe0
--- /dev/null
+++ b/benchmarks/nested/queries/q18.sql
@@ -0,0 +1,37 @@
+-- CometBench-H nested query 18 derived from TPC-H query 18 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ customer_data.c_name,
+ c_custkey,
+ o_orderkey,
+ o_orderdate,
+ orders_data.o_totalprice,
+ sum(lineitem_data.l_quantity)
+from
+ (select c_custkey, c_nationkey, c_mktsegment, explode(customer_data) as customer_data from customer) as customer,
+ (select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+where
+ o_orderkey in (
+ select
+ l_orderkey
+ from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+ group by
+ l_orderkey having
+ sum(lineitem_data.l_quantity) > 300
+ )
+ and c_custkey = o_custkey
+ and o_orderkey = l_orderkey
+group by
+ customer_data.c_name,
+ c_custkey,
+ o_orderkey,
+ o_orderdate,
+ orders_data.o_totalprice
+order by
+ orders_data.o_totalprice desc,
+ o_orderdate
+limit 100
\ No newline at end of file
diff --git a/benchmarks/nested/queries/q19.sql b/benchmarks/nested/queries/q19.sql
new file mode 100644
index 0000000000..faed3de3c3
--- /dev/null
+++ b/benchmarks/nested/queries/q19.sql
@@ -0,0 +1,39 @@
+-- CometBench-H nested query 19 derived from TPC-H query 19 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ sum(lineitem_data.l_extendedprice* (1 - lineitem_data.l_discount)) as revenue
+from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem,
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part
+where
+ (
+ p_partkey = l_partkey
+ and p_brand = 'Brand#12'
+ and part_data.p_container in ('SM CASE', 'SM BOX', 'SM PACK', 'SM PKG')
+ and lineitem_data.l_quantity >= 1 and lineitem_data.l_quantity <= 1 + 10
+ and part_data.p_size between 1 and 5
+ and lineitem_data.l_shipmode in ('AIR', 'AIR REG')
+ and lineitem_data.l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ or
+ (
+ p_partkey = l_partkey
+ and p_brand = 'Brand#23'
+ and part_data.p_container in ('MED BAG', 'MED BOX', 'MED PKG', 'MED PACK')
+ and lineitem_data.l_quantity >= 10 and lineitem_data.l_quantity <= 10 + 10
+ and part_data.p_size between 1 and 10
+ and lineitem_data.l_shipmode in ('AIR', 'AIR REG')
+ and lineitem_data.l_shipinstruct = 'DELIVER IN PERSON'
+ )
+ or
+ (
+ p_partkey = l_partkey
+ and p_brand = 'Brand#34'
+ and part_data.p_container in ('LG CASE', 'LG BOX', 'LG PACK', 'LG PKG')
+ and lineitem_data.l_quantity >= 20 and lineitem_data.l_quantity <= 20 + 10
+ and part_data.p_size between 1 and 15
+ and lineitem_data.l_shipmode in ('AIR', 'AIR REG')
+ and lineitem_data.l_shipinstruct = 'DELIVER IN PERSON'
+ )
diff --git a/benchmarks/nested/queries/q20.sql b/benchmarks/nested/queries/q20.sql
new file mode 100644
index 0000000000..9507d3f244
--- /dev/null
+++ b/benchmarks/nested/queries/q20.sql
@@ -0,0 +1,41 @@
+-- CometBench-H nested query 20 derived from TPC-H query 20 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ supplier_data.s_name,
+ supplier_data.s_address
+from
+ (select s_suppkey, s_nationkey, explode(supplier_data) as supplier_data from supplier) as supplier,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
+where
+ s_suppkey in (
+ select
+ ps_suppkey
+ from
+ (select ps_partkey, ps_suppkey, explode(partsupp_data) as partsupp_data from partsupp) as partsupp
+ where
+ ps_partkey in (
+ select
+ p_partkey
+ from
+ (select p_partkey, p_brand, explode(part_data) as part_data from part) as part
+ where
+ part_data.p_name like 'forest%'
+ )
+ and partsupp_data.ps_availqty > (
+ select
+ 0.5 * sum(lineitem_data.l_quantity)
+ from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as lineitem
+ where
+ l_partkey = ps_partkey
+ and l_suppkey = ps_suppkey
+ and l_shipdate >= date '1994-01-01'
+ and l_shipdate < date '1994-01-01' + interval '1' year
+ )
+ )
+ and s_nationkey = n_nationkey
+ and nation_data.n_name = 'CANADA'
+order by
+ supplier_data.s_name
diff --git a/benchmarks/nested/queries/q21.sql b/benchmarks/nested/queries/q21.sql
new file mode 100644
index 0000000000..ec0f705b97
--- /dev/null
+++ b/benchmarks/nested/queries/q21.sql
@@ -0,0 +1,44 @@
+-- CometBench-H nested query 21 derived from TPC-H query 21 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ supplier_data.s_name,
+ count(*) as numwait
+from
+ (select s_suppkey, s_nationkey, explode(supplier_data) as supplier_data from supplier) as supplier,
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as l1,
+ (select o_orderkey, o_custkey, o_orderdate, explode(orders_data) as orders_data from orders) as orders,
+ (select n_nationkey, n_regionkey, explode(nation_data) as nation_data from nation) as nation
+where
+ s_suppkey = l1.l_suppkey
+ and o_orderkey = l1.l_orderkey
+ and orders_data.o_orderstatus = 'F'
+ and l1.lineitem_data.l_receiptdate > l1.lineitem_data.l_commitdate
+ and exists (
+ select
+ *
+ from
+ lineitem l2
+ where
+ l2.l_orderkey = l1.l_orderkey
+ and l2.l_suppkey <> l1.l_suppkey
+ )
+ and not exists (
+ select
+ *
+ from
+ (select l_orderkey, l_partkey, l_suppkey, l_shipdate, explode(lineitem_data) as lineitem_data from lineitem) as l3
+ where
+ l3.l_orderkey = l1.l_orderkey
+ and l3.l_suppkey <> l1.l_suppkey
+ and l3.lineitem_data.l_receiptdate > l3.lineitem_data.l_commitdate
+ )
+ and s_nationkey = n_nationkey
+ and nation_data.n_name = 'SAUDI ARABIA'
+group by
+ supplier_data.s_name
+order by
+ numwait desc,
+ supplier_data.s_name
+limit 100
\ No newline at end of file
diff --git a/benchmarks/nested/queries/q22.sql b/benchmarks/nested/queries/q22.sql
new file mode 100644
index 0000000000..4257da903d
--- /dev/null
+++ b/benchmarks/nested/queries/q22.sql
@@ -0,0 +1,41 @@
+-- CometBench-H nested query 22 derived from TPC-H query 22 under the terms of the TPC Fair Use Policy.
+-- TPC-H queries are Copyright 1993-2022 Transaction Processing Performance Council.
+-- using default substitutions
+
+select
+ cntrycode,
+ count(*) as numcust,
+ sum(c_acctbal) as totacctbal
+from
+ (
+ select
+ substring(customer_data.c_phone, 1, 2) as cntrycode,
+ customer_data.c_acctbal as c_acctbal
+ from
+ (select c_custkey, c_nationkey, c_mktsegment, explode(customer_data) as customer_data from customer) as customer
+ where
+ substring(customer_data.c_phone, 1, 2) in
+ ('13', '31', '23', '29', '30', '18', '17')
+ and customer_data.c_acctbal > (
+ select
+ avg(customer_data.c_acctbal)
+ from
+ (select c_custkey, c_nationkey, c_mktsegment, explode(customer_data) as customer_data from customer) as customer
+ where
+ customer_data.c_acctbal > 0.00
+ and substring(customer_data.c_phone, 1, 2) in
+ ('13', '31', '23', '29', '30', '18', '17')
+ )
+ and not exists (
+ select
+ *
+ from
+ orders
+ where
+ o_custkey = c_custkey
+ )
+ ) as custsale
+group by
+ cntrycode
+order by
+ cntrycode
diff --git a/benchmarks/nested/scripts/nest_tpch.scala b/benchmarks/nested/scripts/nest_tpch.scala
new file mode 100644
index 0000000000..458595aba0
--- /dev/null
+++ b/benchmarks/nested/scripts/nest_tpch.scala
@@ -0,0 +1,279 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Reads already generated, flat TPC-H tables and rewrites them into a nested layout.
+//
+// For each table, join keys (columns whose name ends in "key") and the partition columns
+// (p_brand, c_mktsegment, o_orderdate, l_shipdate) stay at the top level. Every other column
+// is folded into a single-element array<struct> named `<table>_data`:
+//
+// lineitem(l_orderkey, l_partkey, l_suppkey, l_shipdate,
+// lineitem_data: array<struct<l_linenumber, l_quantity, ..., l_comment>>)
+//
+// The array always holds exactly one element, so exploding it never changes the row count
+// and the nested queries return the same answers as the flat TPC-H queries.
+//
+// Source columns are cast to the types below, so the output schema is the same whichever
+// generator produced the flat data.
+//
+// Source and target can each be Parquet or Iceberg:
+//
+// parquet:<path> one table per <path>/<table> directory (or <path>/<table>.parquet)
+// iceberg:<catalog>.<db> tables <catalog>.<db>.<table>, catalogs configured the usual way
+// with spark.sql.catalog.<name>.* settings
+//
+// Usage, from benchmarks/nested:
+//
+// spark-shell -i scripts/nest_tpch.scala \
+// --conf spark.nesttpch.source=parquet:/data/tpch/sf100 \
+// --conf spark.nesttpch.target=parquet:/data/tpch-nested/sf100
+//
+// Configuration:
+// spark.nesttpch.source Required. Flat tables, parquet:<path> or iceberg:<catalog>.<db>
+// spark.nesttpch.target Required. Where to write the nested tables, same forms
+// spark.nesttpch.tables Comma separated subset of tables (default: all, smallest first)
+// spark.nesttpch.partitioned Partition part, customer, orders and lineitem by p_brand,
+// c_mktsegment, o_orderdate and l_shipdate (default true)
+// spark.nesttpch.overwrite Replace existing target tables (default true)
+// spark.nesttpch.verify Compare source and target row counts (default true)
+// spark.nesttpch.dryRun Print the target DDL and exit without reading or writing data
+// (default false)
+
+import org.apache.hadoop.fs.Path
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.functions.{array, col, struct}
+
+object NestTpch {
+
+ final case class Column(name: String, sqlType: String)
+
+ final case class Table(
+ name: String,
+ topLevel: List[Column],
+ nested: List[Column],
+ partitionColumn: Option[String] = None) {
+ def structName: String = name + "_data"
+ def columns: List[Column] = topLevel ++ nested
+ }
+
+ private def c(name: String, sqlType: String): Column = Column(name, sqlType)
+
+ private val Money = "DECIMAL(12,2)"
+
+ // Smallest first, so a bad configuration fails in seconds rather than hours.
+ val Tables: List[Table] = List(
+ Table("region",
+ List(c("r_regionkey", "BIGINT")),
+ List(c("r_name", "STRING"), c("r_comment", "STRING"))),
+ Table("nation",
+ List(c("n_nationkey", "BIGINT"), c("n_regionkey", "BIGINT")),
+ List(c("n_name", "STRING"), c("n_comment", "STRING"))),
+ Table("supplier",
+ List(c("s_suppkey", "BIGINT"), c("s_nationkey", "BIGINT")),
+ List(c("s_name", "STRING"), c("s_address", "STRING"), c("s_phone", "STRING"),
+ c("s_acctbal", Money), c("s_comment", "STRING"))),
+ Table("part",
+ List(c("p_partkey", "BIGINT"), c("p_brand", "STRING")),
+ List(c("p_name", "STRING"), c("p_mfgr", "STRING"), c("p_type", "STRING"),
+ c("p_size", "BIGINT"), c("p_container", "STRING"), c("p_retailprice", Money),
+ c("p_comment", "STRING")),
+ Some("p_brand")),
+ Table("partsupp",
+ List(c("ps_partkey", "BIGINT"), c("ps_suppkey", "BIGINT")),
+ List(c("ps_availqty", "INT"), c("ps_supplycost", Money), c("ps_comment", "STRING"))),
+ Table("customer",
+ List(c("c_custkey", "BIGINT"), c("c_nationkey", "BIGINT"), c("c_mktsegment", "STRING")),
+ List(c("c_name", "STRING"), c("c_address", "STRING"), c("c_phone", "STRING"),
+ c("c_acctbal", Money), c("c_comment", "STRING")),
+ Some("c_mktsegment")),
+ Table("orders",
+ List(c("o_orderkey", "BIGINT"), c("o_custkey", "BIGINT"), c("o_orderdate", "DATE")),
+ List(c("o_orderstatus", "STRING"), c("o_totalprice", Money), c("o_orderpriority", "STRING"),
+ c("o_clerk", "STRING"), c("o_shippriority", "INT"), c("o_comment", "STRING")),
+ Some("o_orderdate")),
+ Table("lineitem",
+ List(c("l_orderkey", "BIGINT"), c("l_partkey", "BIGINT"), c("l_suppkey", "BIGINT"),
+ c("l_shipdate", "DATE")),
+ List(c("l_linenumber", "INT"), c("l_quantity", Money), c("l_extendedprice", Money),
+ c("l_discount", Money), c("l_tax", Money), c("l_returnflag", "STRING"),
+ c("l_linestatus", "STRING"), c("l_commitdate", "DATE"), c("l_receiptdate", "DATE"),
+ c("l_shipinstruct", "STRING"), c("l_shipmode", "STRING"), c("l_comment", "STRING")),
+ Some("l_shipdate")))
+
+ // Applied to both Parquet and Iceberg targets.
+ val RowGroupBytes: Long = 512L * 1024 * 1024
+ val Compression = "zstd"
+
+ // ---------------------------------------------------------------- locations
+
+ sealed trait Location {
+ def read(spark: SparkSession, table: String): DataFrame
+ }
+
+ final case class ParquetLocation(path: String) extends Location {
+ def tablePath(table: String): String = path.stripSuffix("/") + "/" + table
+
+ def read(spark: SparkSession, table: String): DataFrame = {
+ // Directory per table is the common layout. Some generators write <table>.parquet.
+ val dir = new Path(tablePath(table))
+ val fs = dir.getFileSystem(spark.sparkContext.hadoopConfiguration)
+ val input = if (fs.exists(dir)) dir.toString else dir.toString + ".parquet"
+ spark.read.parquet(input)
+ }
+
+ override def toString: String = s"parquet:$path"
+ }
+
+ final case class IcebergLocation(namespace: String) extends Location {
+ def tableName(table: String): String = s"$namespace.$table"
+ def read(spark: SparkSession, table: String): DataFrame = spark.table(tableName(table))
+ override def toString: String = s"iceberg:$namespace"
+ }
+
+ def parseLocation(spec: String): Location = spec.split(":", 2) match {
+ case Array("parquet", path) if path.nonEmpty => ParquetLocation(path)
+ case Array("iceberg", ns) if ns.contains(".") => IcebergLocation(ns)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"Bad location '$spec'. Expected parquet:<path> or iceberg:<catalog>.<database>")
+ }
+
+ // ---------------------------------------------------------------- options
+
+ final case class Options(
+ source: Location,
+ target: Location,
+ tables: List[Table],
+ partitioned: Boolean,
+ overwrite: Boolean,
+ verify: Boolean,
+ dryRun: Boolean)
+
+ def parseOptions(spark: SparkSession): Options = {
+ def get(key: String): Option[String] =
+ spark.conf.getOption(s"spark.nesttpch.$key").map(_.trim).filter(_.nonEmpty)
+ def required(key: String): String =
+ get(key).getOrElse(throw new IllegalArgumentException(s"spark.nesttpch.$key is required"))
+ def flag(key: String, default: Boolean): Boolean = get(key).map(_.toBoolean).getOrElse(default)
+
+ val names = Tables.map(_.name)
+ val wanted = get("tables").toList.flatMap(_.split(',')).map(_.trim.toLowerCase).filter(_.nonEmpty)
+ wanted.foreach { t =>
+ require(names.contains(t), s"Unknown table '$t'. TPC-H tables: ${names.mkString(", ")}")
+ }
+ Options(
+ source = parseLocation(required("source")),
+ target = parseLocation(required("target")),
+ tables = if (wanted.isEmpty) Tables else Tables.filter(t => wanted.contains(t.name)),
+ partitioned = flag("partitioned", default = true),
+ overwrite = flag("overwrite", default = true),
+ verify = flag("verify", default = true),
+ dryRun = flag("dryRun", default = false))
+ }
+
+ // ---------------------------------------------------------------- converting
+
+ /** Iceberg DDL for the nested table. Also printed for Parquet targets, as documentation. */
+ def createTable(t: Table, name: String, partitioned: Boolean): String = {
+ val width = (t.structName :: t.topLevel.map(_.name)).map(_.length).max
+ def pad(s: String): String = s.padTo(width, ' ')
+ val top = t.topLevel.map(col => s" ${pad(col.name)} ${col.sqlType},")
+ val inner = t.nested.map(col => s" ${col.name}: ${col.sqlType}").mkString(",\n")
+ val partitionBy = t.partitionColumn.filter(_ => partitioned)
+ val props = List(
+ Some("write.parquet.row-group-size-bytes" -> RowGroupBytes.toString),
+ Some("write.parquet.compression-codec" -> Compression),
+ // Cluster rows by partition before writing, so each partition gets few, large files.
+ partitionBy.map(_ => "write.distribution-mode" -> "hash")).flatten
+ (List(s"CREATE TABLE $name (") ++ top ++
+ List(s" ${pad(t.structName)} ARRAY<STRUCT<", inner + ">>", ")", "USING iceberg") ++
+ partitionBy.map(p => s"PARTITIONED BY ($p)").toList ++
+ List("TBLPROPERTIES (", props.map { case (k, v) => s" '$k'='$v'" }.mkString(",\n"), ")"))
+ .mkString("\n")
+ }
+
+ /** Flat source table to nested rows, with every column cast to the target type. */
+ def nest(t: Table, df: DataFrame): DataFrame = {
+ val available = df.columns.map(_.toLowerCase).toSet
+ val missing = t.columns.map(_.name).filterNot(available.contains)
+ require(missing.isEmpty, s"[${t.name}] source has no column ${missing.mkString(", ")}")
+
+ def typed(c: Column) = col(c.name).cast(c.sqlType).as(c.name)
+ // Single-element array, so exploding it never changes the row count. The point is to put
+ // the columns behind an array<struct> access path, not to fan rows out.
+ val projection = t.topLevel.map(typed) :+ array(struct(t.nested.map(typed): _*)).as(t.structName)
+ df.select(projection: _*)
+ }
+
+ def convert(spark: SparkSession, t: Table, opts: Options): Unit = {
+ val partitionBy = t.partitionColumn.filter(_ => opts.partitioned).toList
+ val ddlName = opts.target match {
+ case i: IcebergLocation => i.tableName(t.name)
+ case p: ParquetLocation => s"parquet.`${p.tablePath(t.name)}`"
+ }
+ println(s"[${t.name}] ${opts.source} -> ${opts.target}")
+ println(createTable(t, ddlName, opts.partitioned))
+ if (opts.dryRun) return
+
+ val source = opts.source.read(spark, t.name)
+ val nested = nest(t, source)
+
+ opts.target match {
+ case p: ParquetLocation =>
+ // One shuffle so that each partition directory gets few files rather than one per task.
+ val clustered = if (partitionBy.isEmpty) nested else nested.repartition(partitionBy.map(col): _*)
+ clustered.write
+ .mode(if (opts.overwrite) "overwrite" else "errorifexists")
+ .option("compression", Compression)
+ .option("parquet.block.size", RowGroupBytes.toString)
+ .partitionBy(partitionBy: _*)
+ .parquet(p.tablePath(t.name))
+
+ case i: IcebergLocation =>
+ val name = i.tableName(t.name)
+ spark.sql(s"CREATE NAMESPACE IF NOT EXISTS ${i.namespace}")
+ if (opts.overwrite) spark.sql(s"DROP TABLE IF EXISTS $name")
+ spark.sql(createTable(t, name, opts.partitioned))
+ nested.writeTo(name).append()
+ }
+
+ if (opts.verify) {
+ val sourceRows = source.count()
+ val targetRows = opts.target.read(spark, t.name).count()
+ require(sourceRows == targetRows,
+ s"[${t.name}] row count mismatch: source=$sourceRows target=$targetRows")
+ println(s"[${t.name}] done, $targetRows rows")
+ } else {
+ println(s"[${t.name}] done")
+ }
+ }
+
+ def main(spark: SparkSession): Unit = {
+ val opts = parseOptions(spark)
+ println(s"Source: ${opts.source}")
+ println(s"Target: ${opts.target}")
+ println(s"Tables: ${opts.tables.map(_.name).mkString(", ")}")
+ opts.tables.foreach(convert(spark, _, opts))
+ if (opts.dryRun) println("Dry run: nothing was read or written")
+ else println(s"Converted ${opts.tables.size} table(s) into ${opts.target}")
+ }
+}
+
+try { NestTpch.main(spark); sys.exit(0) }
+catch { case e: Throwable => e.printStackTrace(); sys.exit(1) }
diff --git a/dev/release/rat_exclude_files.txt b/dev/release/rat_exclude_files.txt
index 30e822a6e4..5c660b802c 100644
--- a/dev/release/rat_exclude_files.txt
+++ b/dev/release/rat_exclude_files.txt
@@ -33,3 +33,4 @@ spark/src/test/resources/test-data/*.ndjson
spark/src/test/resources/abfs-auth-parity/*.json
spark/inspections/CometTPC*results.txt
benchmarks/tpc/queries/**/*.sql
+benchmarks/nested/queries/*.sql
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org