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:19 UTC
(datafusion-comet) 01/01: feat: add nested benchmark derived from TPC-H (#7044)
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