Skip to content

Commit 1671cb3

Browse files
robert3005robert3005
authored
Use same tpchgen-rs for sqllogic test and bench (#10656)
switches from duckdb to tpcdsgen-rs
1 parent 8374e7a commit 1671cb3

490 files changed

Lines changed: 19265 additions & 37599 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎Cargo.lock‎

Lines changed: 8 additions & 8 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -279,10 +279,14 @@ tokio = { version = "1.52" }
279279
tokio-stream = "0.1.17"
280280
tokio-util = "0.7.17"
281281
toml = "1.0"
282-
tpcdsgen = "0.1.0-alpha.1"
283-
tpcdsgen-arrow = "0.1.0-alpha.1"
284-
tpchgen = "3.0.0"
285-
tpchgen-arrow = "3.0.0"
282+
tpcdsgen = "4.0.0"
283+
tpcdsgen-arrow = { version = "4.0.0", default-features = false, features = [
284+
"arrow_59",
285+
] }
286+
tpchgen = "4.0.0"
287+
tpchgen-arrow = { version = "4.0.0", default-features = false, features = [
288+
"arrow_59",
289+
] }
286290
tracing = { version = "0.1.41", default-features = false }
287291
tracing-perfetto = "0.1.5"
288292
tracing-subscriber = "0.3"

‎vortex-bench/sql/tpch/README.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,8 @@ and date columns.
77

88
The queries in this directory (`q1.sql` ... `q22.sql`) are the standard TPC-H queries.
99
Data is generated deterministically by [`tpchgen`](../../src/tpch/tpchgen.rs); the benchmark
10-
harness lives in [`src/tpch`](../../src/tpch).
10+
harness lives in [`src/tpch`](../../src/tpch). The SQL logic tests in `vortex-sqllogictest`
11+
generate their scale factor 0.1 fixtures through the same `data-gen` binary.
1112

1213
## CI variants
1314

‎vortex-bench/src/tpcds/tpcdsgen.rs‎

Lines changed: 9 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -7,16 +7,9 @@ use std::fs;
77
use std::fs::File;
88
use std::path::Path;
99
use std::path::PathBuf;
10-
use std::sync::Arc;
1110

1211
use anyhow::Result;
13-
use arrow_array::RecordBatch;
14-
use arrow_cast::CastOptions;
15-
use arrow_cast::cast_with_options;
16-
use arrow_schema::DataType;
17-
use arrow_schema::Field;
18-
use arrow_schema::Schema;
19-
use arrow_schema::SchemaRef;
12+
use arrow_array::RecordBatchReader;
2013
use futures::StreamExt;
2114
use futures::stream;
2215
use parquet::arrow::ArrowWriter;
@@ -37,7 +30,6 @@ use tpcdsgen_arrow::InventoryArrow;
3730
use tpcdsgen_arrow::ItemArrow;
3831
use tpcdsgen_arrow::PromotionArrow;
3932
use tpcdsgen_arrow::ReasonArrow;
40-
use tpcdsgen_arrow::RecordBatchIterator;
4133
use tpcdsgen_arrow::ShipModeArrow;
4234
use tpcdsgen_arrow::StoreArrow;
4335
use tpcdsgen_arrow::StoreReturnsArrow;
@@ -62,16 +54,15 @@ const ROW_GROUP_SIZE: usize = 122_880;
6254

6355
/// Generate every TPC-DS table as a single Parquet file per table under `base_dir/parquet`.
6456
///
65-
/// The files use the column names and types of the TPC-DS specification (and of DuckDB's
66-
/// `dsdgen`). Each table is generated on its own blocking thread. Already generated tables are
67-
/// skipped.
57+
/// The files hold the same data and schema as `tpcgen-cli tpcds parquet` of the same `tpcdsgen`
58+
/// version, which generates the SQL logic test data. Each table is generated on its own blocking
59+
/// thread. Already generated tables are skipped.
6860
pub async fn generate_tpcds(base_dir: PathBuf, scale_factor: String) -> Result<PathBuf> {
6961
let output_dir = base_dir.join(Format::Parquet.name());
7062
fs::create_dir_all(&output_dir)?;
7163

7264
let scale_factor: f64 = scale_factor.parse()?;
73-
// Follow the C `dsdgen` reference rather than the Trino Java port. The generated values
74-
// still differ from DuckDB's bundled `dsdgen`, which is a different version of the C tool.
65+
// Follow the C `dsdgen` reference rather than the Trino Java port, like `tpcgen-cli` does.
7566
let session = SessionBuilder::new()
7667
.with_scale_factor(scale_factor)
7768
.with_compat_mode(CompatMode::C)
@@ -125,7 +116,7 @@ const TABLES: [&str; 24] = [
125116
"web_site",
126117
];
127118

128-
fn table_batches(table: &str, session: Session) -> Box<dyn RecordBatchIterator> {
119+
fn table_batches(table: &str, session: Session) -> Box<dyn RecordBatchReader + Send> {
129120
match table {
130121
"call_center" => Box::new(CallCenterArrow::new(session).with_batch_size(BATCH_SIZE)),
131122
"catalog_page" => Box::new(CatalogPageArrow::new(session).with_batch_size(BATCH_SIZE)),
@@ -163,67 +154,14 @@ fn table_batches(table: &str, session: Session) -> Box<dyn RecordBatchIterator>
163154
}
164155
}
165156

166-
fn write_parquet(path: &Path, batches: Box<dyn RecordBatchIterator>) -> Result<()> {
167-
let schema = spec_schema(batches.schema());
157+
fn write_parquet(path: &Path, batches: Box<dyn RecordBatchReader + Send>) -> Result<()> {
168158
let props = parquet_writer_properties()
169159
.set_max_row_group_row_count(Some(ROW_GROUP_SIZE))
170160
.build();
171-
let mut writer = ArrowWriter::try_new(File::create(path)?, Arc::clone(&schema), Some(props))?;
172-
// Fail on values that do not fit the narrower spec types instead of silently nulling them.
173-
let options = CastOptions {
174-
safe: false,
175-
..Default::default()
176-
};
161+
let mut writer = ArrowWriter::try_new(File::create(path)?, batches.schema(), Some(props))?;
177162
for batch in batches {
178-
let columns = batch
179-
.columns()
180-
.iter()
181-
.zip(schema.fields())
182-
.map(|(column, field)| cast_with_options(column, field.data_type(), &options))
183-
.collect::<Result<Vec<_>, _>>()?;
184-
writer.write(&RecordBatch::try_new(Arc::clone(&schema), columns)?)?;
163+
writer.write(&batch?)?;
185164
}
186165
writer.close()?;
187166
Ok(())
188167
}
189-
190-
/// Columns that `tpcdsgen-arrow` names after the Trino Java port instead of the specification.
191-
const RENAMED_COLUMNS: [(&str, &str); 5] = [
192-
("cr_return_amount_inc_tax", "cr_return_amt_inc_tax"),
193-
("ib_income_band_id", "ib_income_band_sk"),
194-
("r_reason_description", "r_reason_desc"),
195-
("s_tax_precentage", "s_tax_percentage"),
196-
("wr_store_credit", "wr_account_credit"),
197-
];
198-
199-
/// Integer columns that DuckDB declares as `INTEGER`. All other integers are `BIGINT`.
200-
const INT32_COLUMNS: [&str; 6] = [
201-
"c_last_review_date_sk",
202-
"cd_dep_college_count",
203-
"hd_vehicle_count",
204-
"ib_upper_bound",
205-
"inv_quantity_on_hand",
206-
"wp_max_ad_count",
207-
];
208-
209-
/// Map a `tpcdsgen-arrow` schema onto the column names and types of the TPC-DS specification.
210-
fn spec_schema(schema: &SchemaRef) -> SchemaRef {
211-
let fields = schema.fields().iter().map(|field| {
212-
let name = RENAMED_COLUMNS
213-
.iter()
214-
.find(|(from, _)| *from == field.name())
215-
.map_or_else(|| field.name().as_str(), |(_, to)| to);
216-
let data_type = match field.data_type() {
217-
_ if name.ends_with("_street_number") => DataType::Utf8,
218-
_ if name.ends_with("_gmt_offset") || name.ends_with("_tax_percentage") => {
219-
DataType::Decimal128(5, 2)
220-
}
221-
DataType::Decimal128(..) if name == "p_cost" => DataType::Decimal128(15, 2),
222-
DataType::Decimal128(..) => DataType::Decimal128(7, 2),
223-
DataType::Int32 if !INT32_COLUMNS.contains(&name) => DataType::Int64,
224-
data_type => data_type.clone(),
225-
};
226-
Field::new(name, data_type, true)
227-
});
228-
Arc::new(Schema::new(fields.collect::<Vec<_>>()))
229-
}

‎vortex-bench/src/tpch/tpchgen.rs‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ use std::sync::Arc;
99
use anyhow::Result;
1010
use anyhow::anyhow;
1111
use arrow_array::RecordBatch;
12+
use arrow_array::RecordBatchReader;
1213
use arrow_schema::SchemaRef;
1314
use futures::FutureExt;
1415
use futures::StreamExt;
@@ -27,7 +28,6 @@ use tpchgen::generators::PartGenerator;
2728
use tpchgen::generators::PartSuppGenerator;
2829
use tpchgen::generators::RegionGenerator;
2930
use tpchgen::generators::SupplierGenerator;
30-
use tpchgen_arrow::RecordBatchIterator;
3131
use tracing::info;
3232
use vortex::array::ArrayRef;
3333
use vortex::array::stream::ArrayStreamAdapter;
@@ -193,7 +193,7 @@ fn generate_table_file(
193193
let iter = create_batch_iterator(generator, &options)?;
194194

195195
// Create generator and process batches in streaming fashion
196-
let schema = Arc::clone(iter.schema());
196+
let schema = iter.schema();
197197

198198
// Create writer based on format
199199
let mut writer: Box<dyn FileWriter + Send> = match write_format {
@@ -212,7 +212,7 @@ fn generate_table_file(
212212
};
213213

214214
for batch in iter {
215-
writer.write_batch(&batch).await?;
215+
writer.write_batch(&batch?).await?;
216216
}
217217

218218
writer.finalize().await?;
@@ -230,13 +230,13 @@ fn generate_table_file(
230230
fn create_batch_iterator(
231231
generator: TableGenerator,
232232
options: &TpchGenOptions,
233-
) -> Result<Box<dyn RecordBatchIterator>> {
233+
) -> Result<Box<dyn RecordBatchReader + Send>> {
234234
let scale_factor = options.scale_factor.parse::<f64>()?;
235235
let part = 1;
236236
let num_parts = 1;
237237
let batch_size = options.batch_size;
238238

239-
let iterator: Box<dyn RecordBatchIterator> = match generator {
239+
let iterator: Box<dyn RecordBatchReader + Send> = match generator {
240240
TableGenerator::Nation => Box::new(
241241
tpchgen_arrow::NationArrow::new(NationGenerator::new(scale_factor, part, num_parts))
242242
.with_batch_size(batch_size),

‎vortex-sqllogictest/README.md‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,8 @@ cargo test -p vortex-sqllogictest --test sqllogictests
2525
`./vortex-sqllogictest/slt/generate_data.sh tpch`, `... tpcds` or `... clickbench`. Run it with
2626
`--help` to list the datasets. It runs each dataset's own script, `slt/tpch/generate_data.sh`,
2727
`slt/tpcds/generate_data.sh` or `slt/clickbench/generate_data.sh`, which can also be run directly.
28-
TPC-DS is generated by DuckDB's `tpcds` extension through `uvx`.
28+
TPC-H is generated by `vortex-bench`'s `data-gen` binary, the same generator the benchmarks use,
29+
and TPC-DS by `tpcgen-cli` through `uvx`.
2930

3031
The generated Vortex and Parquet data lives under `slt/tpch/data/`, `slt/tpcds/data/` and
3132
`slt/clickbench/data/` (git-ignored). Both formats are required; regenerate older fixtures if they
@@ -37,11 +38,18 @@ working directory accordingly.
3738
TPC-H scripts live under `slt/tpch/datafusion/` and `slt/tpch/duckdb/`, TPC-DS scripts under
3839
`slt/tpcds/datafusion/` and `slt/tpcds/duckdb/`, ClickBench scripts under
3940
`slt/clickbench/datafusion/` and `slt/clickbench/duckdb/`. Each engine has its own
40-
`create.slt.no`, `results/q*.slt.no` (`q1` to `q22` for TPC-H, `q1` to `q99` for TPC-DS, `q0` to
41-
`q42` for ClickBench, matching the upstream numbering), and `drop.slt.no`. Its
42-
`tpch.slt`/`tpcds.slt`/`clickbench.slt` runs these against Vortex and asserts EXPLAIN output from
43-
the matching `plans/q*.slt.no`. Its `parquet.slt` runs the same queries against the original
44-
Parquet fixtures and checks the same expected results.
41+
`create.slt.no` and `drop.slt.no`. The query results, `results/q*.slt.no` (`q1` to `q22` for
42+
TPC-H, `q1` to `q99` for TPC-DS, `q0` to `q42` for ClickBench, matching the upstream numbering),
43+
are shared by both engines in `slt/tpch/results/` and `slt/tpcds/results/`. ClickBench keeps one
44+
`results/` directory for each engine. Each engine's `tpch.slt`/`tpcds.slt`/`clickbench.slt` runs
45+
the queries against Vortex and asserts EXPLAIN output from the matching `plans/q*.slt.no`. Its
46+
`parquet.slt` runs the same queries against the original Parquet fixtures and checks the same
47+
expected results.
48+
49+
DataFusion truncates decimal division and `avg` to the scale of the result type, while DuckDB
50+
calculates them as `DOUBLE`. Thus the DuckDB validator accepts a value with more fractional digits
51+
when it truncates to the expected value. Use `onlyif datafusion`/`onlyif duckdb` records when the
52+
engines differ more than that, as TPC-DS q66 does.
4553
DataFusion uses external tables; DuckDB uses views over files. The `FILE_FORMAT` substitution
4654
variable selects the format in each engine's table setup.
4755

‎vortex-sqllogictest/slt/generate_data.sh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010
#
1111
# Datasets:
1212
# tpch TPC-H at scale factor 0.1.
13-
# tpcds TPC-DS at scale factor 0.1 from DuckDB's tpcds extension.
13+
# tpcds TPC-DS at scale factor 0.1.
1414
# clickbench One shard (~1M rows) of the partitioned ClickBench `hits` table.
1515
#
1616
# With no arguments every dataset is generated.

‎vortex-sqllogictest/slt/tpcds/datafusion/parquet.slt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,5 +7,5 @@ let FILE_FORMAT
77
SELECT 'parquet';
88

99
include ./create.slt.no
10-
include ./results/*.slt.no
10+
include ../results/*.slt.no
1111
include ./drop.slt.no

‎vortex-sqllogictest/slt/tpcds/datafusion/plans/q1.slt.no‎

Lines changed: 20 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -43,8 +43,8 @@ logical_plan
4343
13)------------------------Inner Join: store_returns.sr_returned_date_sk = date_dim.d_date_sk
4444
14)--------------------------TableScan: store_returns projection=[sr_returned_date_sk, sr_customer_sk, sr_store_sk, sr_return_amt]
4545
15)--------------------------Projection: date_dim.d_date_sk
46-
16)----------------------------Filter: date_dim.d_year = Int64(2000)
47-
17)------------------------------TableScan: date_dim projection=[d_date_sk, d_year], partial_filters=[date_dim.d_year = Int64(2000)]
46+
16)----------------------------Filter: date_dim.d_year = Int32(2000)
47+
17)------------------------------TableScan: date_dim projection=[d_date_sk, d_year], partial_filters=[date_dim.d_year = Int32(2000)]
4848
18)--------------Projection: store.s_store_sk
4949
19)----------------Filter: store.s_state = Utf8View("TN")
5050
20)------------------TableScan: store projection=[s_store_sk, s_state], partial_filters=[store.s_state = Utf8View("TN")]
@@ -60,8 +60,8 @@ logical_plan
6060
30)----------------------Inner Join: store_returns.sr_returned_date_sk = date_dim.d_date_sk
6161
31)------------------------TableScan: store_returns projection=[sr_returned_date_sk, sr_customer_sk, sr_store_sk, sr_return_amt]
6262
32)------------------------Projection: date_dim.d_date_sk
63-
33)--------------------------Filter: date_dim.d_year = Int64(2000)
64-
34)----------------------------TableScan: date_dim projection=[d_date_sk, d_year], partial_filters=[date_dim.d_year = Int64(2000)]
63+
33)--------------------------Filter: date_dim.d_year = Int32(2000)
64+
34)----------------------------TableScan: date_dim projection=[d_date_sk, d_year], partial_filters=[date_dim.d_year = Int32(2000)]
6565
physical_plan
6666
01)SortPreservingMergeExec: [c_customer_id@0 ASC NULLS LAST], fetch=100
6767
02)--SortExec: TopK(fetch=100), expr=[c_customer_id@0 ASC NULLS LAST], preserve_partitioning=[true]
@@ -77,17 +77,19 @@ physical_plan
7777
12)--------------------AggregateExec: mode=Partial, gby=[sr_customer_sk@0 as sr_customer_sk, sr_store_sk@1 as sr_store_sk], aggr=[sum(store_returns.sr_return_amt)]
7878
13)----------------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, sr_returned_date_sk@0)], projection=[sr_customer_sk@2, sr_store_sk@3, sr_return_amt@4]
7979
14)------------------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/date_dim.vortex]]}, projection=[d_date_sk], file_type=vortex, predicate: d_year@6 = 2000
80-
15)------------------------DataSourceExec: file_groups={4 groups: [[${WORK_DIR}/store_returns.vortex:<slt:ignore>], [${WORK_DIR}/store_returns.vortex:<slt:ignore>], [${WORK_DIR}/store_returns.vortex:<slt:ignore>], [${WORK_DIR}/store_returns.vortex:<slt:ignore>]]}, projection=[sr_returned_date_sk, sr_customer_sk, sr_store_sk, sr_return_amt], file_type=vortex
81-
16)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
82-
17)------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/customer.vortex]]}, projection=[c_customer_sk, c_customer_id], file_type=vortex
83-
18)------ProjectionExec: expr=[CAST(CAST(avg(ctr2.ctr_total_return)@1 AS Float64) * 1.2 AS Decimal128(30, 15)) as avg(ctr2.ctr_total_return) * Float64(1.2), ctr_store_sk@0 as ctr_store_sk]
84-
19)--------AggregateExec: mode=FinalPartitioned, gby=[ctr_store_sk@0 as ctr_store_sk], aggr=[avg(ctr2.ctr_total_return)]
85-
20)----------RepartitionExec: partitioning=Hash([ctr_store_sk@0], 4), input_partitions=4
86-
21)------------AggregateExec: mode=Partial, gby=[ctr_store_sk@0 as ctr_store_sk], aggr=[avg(ctr2.ctr_total_return)]
87-
22)--------------ProjectionExec: expr=[sr_store_sk@1 as ctr_store_sk, sum(store_returns.sr_return_amt)@2 as ctr_total_return]
88-
23)----------------AggregateExec: mode=FinalPartitioned, gby=[sr_customer_sk@0 as sr_customer_sk, sr_store_sk@1 as sr_store_sk], aggr=[sum(store_returns.sr_return_amt)]
89-
24)------------------RepartitionExec: partitioning=Hash([sr_customer_sk@0, sr_store_sk@1], 4), input_partitions=4
90-
25)--------------------AggregateExec: mode=Partial, gby=[sr_customer_sk@0 as sr_customer_sk, sr_store_sk@1 as sr_store_sk], aggr=[sum(store_returns.sr_return_amt)]
91-
26)----------------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, sr_returned_date_sk@0)], projection=[sr_customer_sk@2, sr_store_sk@3, sr_return_amt@4]
92-
27)------------------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/date_dim.vortex]]}, projection=[d_date_sk], file_type=vortex, predicate: d_year@6 = 2000
93-
28)------------------------DataSourceExec: file_groups={4 groups: [[${WORK_DIR}/store_returns.vortex:<slt:ignore>], [${WORK_DIR}/store_returns.vortex:<slt:ignore>], [${WORK_DIR}/store_returns.vortex:<slt:ignore>], [${WORK_DIR}/store_returns.vortex:<slt:ignore>]]}, projection=[sr_returned_date_sk, sr_customer_sk, sr_store_sk, sr_return_amt], file_type=vortex
80+
15)------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
81+
16)--------------------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/store_returns.vortex]]}, projection=[sr_returned_date_sk, sr_customer_sk, sr_store_sk, sr_return_amt], file_type=vortex
82+
17)----------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
83+
18)------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/customer.vortex]]}, projection=[c_customer_sk, c_customer_id], file_type=vortex
84+
19)------ProjectionExec: expr=[CAST(CAST(avg(ctr2.ctr_total_return)@1 AS Float64) * 1.2 AS Decimal128(30, 15)) as avg(ctr2.ctr_total_return) * Float64(1.2), ctr_store_sk@0 as ctr_store_sk]
85+
20)--------AggregateExec: mode=FinalPartitioned, gby=[ctr_store_sk@0 as ctr_store_sk], aggr=[avg(ctr2.ctr_total_return)]
86+
21)----------RepartitionExec: partitioning=Hash([ctr_store_sk@0], 4), input_partitions=4
87+
22)------------AggregateExec: mode=Partial, gby=[ctr_store_sk@0 as ctr_store_sk], aggr=[avg(ctr2.ctr_total_return)]
88+
23)--------------ProjectionExec: expr=[sr_store_sk@1 as ctr_store_sk, sum(store_returns.sr_return_amt)@2 as ctr_total_return]
89+
24)----------------AggregateExec: mode=FinalPartitioned, gby=[sr_customer_sk@0 as sr_customer_sk, sr_store_sk@1 as sr_store_sk], aggr=[sum(store_returns.sr_return_amt)]
90+
25)------------------RepartitionExec: partitioning=Hash([sr_customer_sk@0, sr_store_sk@1], 4), input_partitions=4
91+
26)--------------------AggregateExec: mode=Partial, gby=[sr_customer_sk@0 as sr_customer_sk, sr_store_sk@1 as sr_store_sk], aggr=[sum(store_returns.sr_return_amt)]
92+
27)----------------------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d_date_sk@0, sr_returned_date_sk@0)], projection=[sr_customer_sk@2, sr_store_sk@3, sr_return_amt@4]
93+
28)------------------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/date_dim.vortex]]}, projection=[d_date_sk], file_type=vortex, predicate: d_year@6 = 2000
94+
29)------------------------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
95+
30)--------------------------DataSourceExec: file_groups={1 group: [[${WORK_DIR}/store_returns.vortex]]}, projection=[sr_returned_date_sk, sr_customer_sk, sr_store_sk, sr_return_amt], file_type=vortex

0 commit comments

Comments
 (0)