This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new f40c80dbf83 [improvement](tools) Run SSB, TPCH and TPCDS benchmarks
with one command (#68278)
f40c80dbf83 is described below
commit f40c80dbf831f30a2da3f4b0fa077eca9b03e511
Author: HappenLee <[email protected]>
AuthorDate: Sat Oct 10 18:03:42 2026 +0800
[improvement](tools) Run SSB, TPCH and TPCDS benchmarks with one command
(#68278)
### What problem does this PR solve?
Issue Number: None
The existing SSB, TPCH and TPCDS tools require generating local data
files and
running separate creation, import, statistics and benchmark commands.
Add one
entry point per suite, backed by a shared runner:
```bash
tools/ssb-tools/bin/run-ssb.sh -s 1 -d ssb_sf1
tools/tpch-tools/bin/run-tpch.sh -s 1 -d tpch_sf1
tools/tpcds-tools/bin/run-tpcds.sh -s 1 -d tpcds_sf1
```
Rows are generated on demand through Doris's Trino connector and
imported with
`INSERT INTO ... SELECT ...`, without intermediate data files or a
separate
Trino server. SSB uses the included native-dbgen plugin. TPCH/TPCDS use
complete
upstream Trino 435 plugin distributions, pinned by SHA-256 and
downloaded during
operator installation. Plugins must be installed on every FE and BE
before use.
The runner executes the original scale-specific table DDL and query SQL:
- SSB prepares its five base tables and supports the original
`lineorder_flat`
workflow through `--mode ssb|flat|both`. Both suites have 13 query
files.
- TPCH prepares its eight base tables and `revenue0` view and runs 22
query files.
- TPCDS prepares its 24 base tables and runs 103 scale-specific query
files,
including the second statements of templates 14, 23, 24 and 39.
Full statistics are collected synchronously before query timing. Every
query
runs three times with SQL and BE query result caches disabled.
`--queries-only`
repeats measurements against already prepared data. Each run preserves
SQL
results, errors, preparation logs, settings and CSV timings in a fresh
directory.
Parallel TPCH/TPCDS tables default to `max(10, SCALE)` generator splits:
10/100/1000/10000 for SF1/100/1000/10000. `--splits COUNT` overrides
that count;
invalid values fail before connecting. TPCH region/nation and TPCDS
tables below
the pinned generator's one-million-record parallelization threshold use
one
split, including with an override. Small tables share a single-split
catalog;
`--splits 1` reuses the main catalog. Per-table effective counts are
logged.
Preparation stderr is saved in `prepare.log`, while failures still print
the
log directory to the terminal. SSB preserves the old generator's ten
lineorder partitions, enables its existing 64-bit path and fixes SSB
random
stream usage-counter bounds. Each SSB scan owns and closes its generator
process.
Imports explicitly project target column names, including TPCH's
reordered
lineitem columns and Trino 435's `p_response_targe` spelling for TPCDS
promotion.
TPCH uses standard prefixed column names and DECIMAL values. A fresh
database
prevents duplicate appends on retries. INSERTs use strict mode with zero
filtered
rows and must return a metadata `status` of `VISIBLE`. SQL text or
labels
containing `COMMITTED` cannot cause a false failure; missing or
non-visible
status stops preparation. Query errors stop the suite and are not
recorded as
successful timings.
---
tools/benchmark/.gitignore | 2 +
tools/benchmark/README.md | 164 +++++++++
tools/benchmark/download-tpc-plugins.sh | 41 +++
tools/benchmark/run-benchmark.sh | 356 +++++++++++++++++++
tools/benchmark/tests/test_run_benchmark.py | 380 +++++++++++++++++++++
tools/ssb-tools/.gitignore | 1 +
tools/ssb-tools/README.md | 92 ++++-
tools/ssb-tools/bin/run-ssb.sh | 21 ++
tools/ssb-tools/ddl/insert-ssb-flat.sql | 30 ++
tools/ssb-tools/trino-ssb/.gitignore | 1 +
tools/ssb-tools/trino-ssb/build.sh | 45 +++
tools/ssb-tools/trino-ssb/dbgen-stream.patch | 62 ++++
tools/ssb-tools/trino-ssb/pom.xml | 104 ++++++
.../org/apache/doris/tools/ssb/SsbConnector.java | 226 ++++++++++++
.../java/org/apache/doris/tools/ssb/SsbPlugin.java | 56 +++
.../org/apache/doris/tools/ssb/SsbRecordSet.java | 198 +++++++++++
.../java/org/apache/doris/tools/ssb/SsbTable.java | 60 ++++
.../META-INF/services/io.trino.spi.Plugin | 18 +
.../apache/doris/tools/ssb/SsbConnectorTest.java | 186 ++++++++++
tools/tpcds-tools/.gitignore | 1 +
tools/tpcds-tools/README.md | 21 ++
tools/tpcds-tools/bin/run-tpcds.sh | 21 ++
tools/tpch-tools/.gitignore | 1 +
tools/tpch-tools/README.md | 21 ++
tools/tpch-tools/bin/run-tpch.sh | 21 ++
25 files changed, 2128 insertions(+), 1 deletion(-)
diff --git a/tools/benchmark/.gitignore b/tools/benchmark/.gitignore
new file mode 100644
index 00000000000..bf5351bd9c2
--- /dev/null
+++ b/tools/benchmark/.gitignore
@@ -0,0 +1,2 @@
+target/
+tests/__pycache__/
diff --git a/tools/benchmark/README.md b/tools/benchmark/README.md
new file mode 100644
index 00000000000..81f2d817083
--- /dev/null
+++ b/tools/benchmark/README.md
@@ -0,0 +1,164 @@
+<!--
+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.
+-->
+
+# Streaming benchmark workflows
+
+The SSB, TPCH, and TPCDS entry points create a new Doris database, generate
rows
+while Doris scans a Trino catalog, import with `INSERT INTO ... SELECT ...`,
+collect statistics synchronously, and execute each existing query three times.
+No intermediate data files or separate Trino server are needed.
+
+| Entry point | Scale factors | Tables | Query files |
+| --- | --- | --- | --- |
+| `../ssb-tools/bin/run-ssb.sh` | 1, 100, 1000 | Five SSB tables and
`lineorder_flat` | 13 SSB + 13 SSB Flat |
+| `../tpch-tools/bin/run-tpch.sh` | 1, 100, 1000, 10000 | Eight TPCH tables
and the `revenue0` view | 22 |
+| `../tpcds-tools/bin/run-tpcds.sh` | 1, 100, 1000, 10000 | 24 TPCDS tables |
103, including the second statements of 14, 23, 24, and 39 |
+
+TPCH and TPCDS use the existing multi-table DDL for the requested scale factor
+and run their original query SQL. SSB supports its original and flat-table
query
+suites, using the existing `lineorder_flat` DDL and joins. All suites collect
+full statistics synchronously before query timing.
+
+## Install the generators once
+
+For SSB, follow [the SSB plugin build instructions](../ssb-tools/README.md).
+For TPCH/TPCDS, download the complete upstream plugin distributions with
`curl`,
+`unzip`, and `sha256sum` installed:
+
+```bash
+./download-tpc-plugins.sh
+cp -r target/trino-tpch-435 target/trino-tpcds-435
/path/to/fe/plugins/trino_plugins/
+cp -r target/trino-tpch-435 target/trino-tpcds-435
/path/to/be/plugins/trino_plugins/
+```
+
+Install both directories on **every FE and BE**, then restart those nodes. The
+script pins Trino 435 to match Doris's Trino SPI and verifies each archive's
+SHA-256 before extracting it. Keep all bundled runtime JARs in each directory;
+copying only the connector JAR is insufficient. Downloads happen only during
+this operator installation step, never from catalog properties.
+
+These plugins use Trino's Java TPCH/TPCDS generators. They are distinct from
the
+native generator versions used by the older file-based scripts; their outputs
+are not asserted to be byte-for-byte identical to those tools. TPCH explicitly
+uses standard prefixed column names and DECIMAL monetary values. Each import
+selects target column names explicitly, so different source and destination
+column orders do not reorder the data. The TPCDS import also maps Trino 435's
`p_response_targe` spelling to the existing Doris
+`promotion.p_response_target` column.
+
+## Run
+
+Set each suite's `conf/doris-cluster.conf`, or pass a separate connection file:
+
+```bash
+../tpch-tools/bin/run-tpch.sh -s 1 -d tpch_sf1 -c /path/to/cluster.conf
+../tpcds-tools/bin/run-tpcds.sh -s 1 -d tpcds_sf1 -c /path/to/cluster.conf
+../ssb-tools/bin/run-ssb.sh -s 1 -d ssb_sf1 -c /path/to/cluster.conf
+```
+
+Parallel TPCH/TPCDS tables default to `max(10, SCALE)` generator splits:
+
+| Scale factor | Default splits for parallel tables |
+| --- | --- |
+| 1 | 10 |
+| 100 | 100 |
+| 1000 | 1000 |
+| 10000 | 10000 |
+
+Use `--splits COUNT` to override this count during preparation, for example:
+
+```bash
+../tpch-tools/bin/run-tpch.sh -s 1000 -d tpch_sf1000 --splits 256
+```
+
+`COUNT` must be an integer from 1 to 2147483647. Small tables always use one
split,
+even when `--splits` is supplied:
+
+- TPCH: `region` and `nation`.
+- TPCDS at every supported scale: `call_center`, `catalog_page`, `date_dim`,
+ `household_demographics`, `income_band`, `item`, `promotion`, `reason`,
+ `ship_mode`, `store`, `time_dim`, `warehouse`, `web_page`, and `web_site`.
+- TPCDS at SF1 also uses one split for `customer`, `customer_address`, and all
+ six sales/returns tables.
+
+The TPCDS choices follow the pinned generator's
+[`Parallel.splitWork`](https://github.com/trinodb/tpcds/blob/1.4/src/main/java/io/trino/tpcds/Parallel.java)
+rule: tables with fewer than one million generator records produce rows only in
+the first chunk. Sales generator records expand into multiple output rows.
+The runner creates a `${suite}_gen_${database}` catalog for parallel tables
and,
+when the selected count exceeds one, a `${suite}_one_${database}` catalog for
+small tables. Both use the same plugin, scale, and column options. Each table's
+effective split count is recorded in `prepare.log`; actual concurrent execution
+depends on cluster resources. `--queries-only` does not recreate or change
these
+catalogs. SSB does not accept `--splits`: it retains the original generator's
ten
+lineorder partitions because changing the partition count changes its random
+data streams. SSB dimension tables use one split each.
+
+Preparation requires a **new database**. An existing database is rejected
before
+any tables are changed, preventing duplicate appends after a partial load.
+Imports use strict mode and zero filtered rows. Each INSERT must return a
+`status` field equal to `VISIBLE`; errors, missing status, or any other status
+(including `COMMITTED`) abort preparation. Failed runs preserve tables and logs
+for inspection. Do not rerun preparation into that database.
+
+To repeat measurements against completed data, use `--queries-only`. For TPCDS,
+pass the same scale factor as the imported data, because query constants differ
+by scale:
+
+```bash
+../tpcds-tools/bin/run-tpcds.sh -s 100 -d tpcds_sf100 --queries-only
+```
+
+SSB also accepts `--mode ssb|flat|both` (default `both`). This option is
rejected
+by TPCH and TPCDS. Existing file-based entry points remain available.
+
+## Results and timing
+
+Each run creates a fresh `<suite>-tools/results/<timestamp>-<pid>/` directory.
+Use `--result-dir /path/to/new-directory` to select another location; existing
+result directories are rejected rather than overwritten.
+
+- `result.csv`: suite, query, first-run milliseconds, two repeated-run times,
and
+ the minimum repeated-run time.
+- `<suite>/q*.cold.out`, `q*.hot1.out`, `q*.hot2.out`: query results, with
matching
+ `.err` files for diagnostics. TPCDS variants have distinct names such as
`q14_1`.
+- `prepare.log`: per-table split counts, DDL, import, and statistics
diagnostics,
+ including preparation errors from stderr.
+- `environment.txt`: Doris version, session variables, and table status.
+
+Timing includes mysql client startup, connection, execution, and result
transfer.
+The `cold_ms` column means the **first execution**; no storage or OS caches are
+flushed. SQL result caching and BE query result caching are disabled in every
+measured session. A query failure stops immediately and never produces a
+successful timing row for that query. These are engineering measurements, not
+certified TPC benchmark results.
+
+## Orchestration tests
+
+From the repository root:
+
+```bash
+python3 -m unittest discover -s tools/benchmark/tests -v
+```
+
+The tests use a fake mysql client to exercise complete suite selection, TPCDS
+split queries, explicit column projection, scale-specific DDL and queries,
+generator split defaults and overrides, cache settings, result isolation,
INSERT
+status checks, and failure paths. They verify that TPCH/TPCDS prepare only
their
+original tables and that SSB retains its flat-table workflow. They do not
replace
+a real Doris import/query run.
diff --git a/tools/benchmark/download-tpc-plugins.sh
b/tools/benchmark/download-tpc-plugins.sh
new file mode 100755
index 00000000000..cce40453708
--- /dev/null
+++ b/tools/benchmark/download-tpc-plugins.sh
@@ -0,0 +1,41 @@
+#!/usr/bin/env bash
+# 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.
+
+set -euo pipefail
+BENCHMARK_ROOT=$(cd "$(dirname "$0")" && pwd)
+cd "${BENCHMARK_ROOT}"
+# Match fe/pom.xml's Trino SPI version. These are the complete upstream plugin
+# distributions, including runtime dependencies, not standalone connector JARs.
+version=435
+mkdir -p target
+for connector in tpch tpcds; do
+ # Both values in the fixed connector list are covered.
+ # shellcheck disable=SC2249
+ case "${connector}" in
+ tpch)
checksum=59d99220b5f47a63606f3bbcd101b7a8e3a160438b6f2fdc361b91badf4695e1 ;;
+ tpcds)
checksum=38b9027d8201cb37ff14d051840aeb41eb29ad11bc89a862d294ce9049ae85e8 ;;
+ esac
+ archive="target/trino-${connector}-${version}.zip"
+ curl --fail --location --retry 3 \
+
"https://repo.maven.apache.org/maven2/io/trino/trino-${connector}/${version}/trino-${connector}-${version}.zip";
\
+ --output "${archive}"
+ echo "${checksum} ${archive}" | sha256sum --check
+ unzip -oq "${archive}" -d target
+done
+echo 'Copy the entire target/trino-tpch-435 and target/trino-tpcds-435
directories'
+echo 'into plugins/trino_plugins/ on every FE and BE, then restart those
nodes.'
diff --git a/tools/benchmark/run-benchmark.sh b/tools/benchmark/run-benchmark.sh
new file mode 100755
index 00000000000..89a3a35827c
--- /dev/null
+++ b/tools/benchmark/run-benchmark.sh
@@ -0,0 +1,356 @@
+#!/usr/bin/env bash
+# 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.
+
+set -Eeuo pipefail
+BENCHMARK_ROOT=$(cd "$(dirname "$0")" && pwd)
+benchmark=$1
+shift
+case "${benchmark}" in
+ ssb | tpch | tpcds) ;;
+ *)
+ echo "Unknown benchmark: ${benchmark}" >&2
+ exit 1
+ ;;
+esac
+SUITE_ROOT="${BENCHMARK_ROOT}/../${benchmark}-tools"
+benchmark_name=${benchmark^^}
+
+usage() {
+ cat <<EOF
+Usage: run-${benchmark}.sh [-s SCALE] [-c CONFIG] [-d DATABASE]
[--queries-only]
+ [--result-dir DIRECTORY] [--mode both|ssb|flat]
[--splits COUNT]
+
+Create tables, generate and import data through the Trino ${benchmark}
connector,
+collect statistics, and run every existing query three times. No data files or
+separate Trino server are needed. SSB can also prepare lineorder_flat.
+Existing table DDL and query SQL are preserved.
+
+SCALE: 1, 100, 1000 (also 10000 for TPCH/TPCDS). Default: 1.
+--mode is only supported by SSB; its default is both.
+--splits controls parallel TPCH/TPCDS tables; its default is max(10, SCALE).
+Small tables use one split, including when --splits is supplied.
+COUNT must be an integer from 1 to 2147483647 and applies during preparation.
+Requires mysql and the generator plugin installed on every Doris FE and BE.
+The default connection configuration is conf/doris-cluster.conf.
+Preparation requires a new database. Use --queries-only against completed data.
+EOF
+}
+
+scale=1
+config="${SUITE_ROOT}/conf/doris-cluster.conf"
+database=''
+queries_only=0
+mode=''
+generator_splits=''
+result_dir="${SUITE_ROOT}/results/$(date +%Y%m%d-%H%M%S)-$$"
+options=$(getopt -o hs:c:d: -l help,queries-only,mode:,result-dir:,splits: --
"$@")
+eval set -- "${options}"
+while true; do
+ case "$1" in
+ -h | --help)
+ usage
+ exit 0
+ ;;
+ -s)
+ scale=$2
+ shift 2
+ ;;
+ -c)
+ config=$2
+ shift 2
+ ;;
+ -d)
+ database=$2
+ shift 2
+ ;;
+ --queries-only)
+ queries_only=1
+ shift
+ ;;
+ --mode)
+ mode=$2
+ shift 2
+ ;;
+ --result-dir)
+ result_dir=$2
+ shift 2
+ ;;
+ --splits)
+ # Both Trino connectors parse the split count as a Java int.
+ if [[ ! $2 =~ ^[1-9][0-9]{0,9}$ ]] || (($2 > 2147483647)); then
+ echo '--splits must be an integer from 1 to 2147483647.' >&2
+ exit 1
+ fi
+ generator_splits=$2
+ shift 2
+ ;;
+ --)
+ shift
+ break
+ ;;
+ *)
+ usage >&2
+ exit 1
+ ;;
+ esac
+done
+if [[ $# -ne 0 || ! ${scale} =~ ^(1|100|1000|10000)$ ]] ||
+ [[ ${benchmark} == ssb && ${scale} == 10000 ]] ||
+ [[ ${benchmark} == ssb && -n ${generator_splits} ]] ||
+ [[ ${benchmark} != ssb && -n ${mode} ]] ||
+ [[ -n ${mode} && ! ${mode} =~ ^(both|ssb|flat)$ ]]; then
+ usage >&2
+ exit 1
+fi
+mode=${mode:-both}
+generator_splits=${generator_splits:-$((scale > 10 ? scale : 10))}
+ddl="${SUITE_ROOT}/ddl/create-${benchmark}-tables-sf${scale}.sql"
+single_split_tables=()
+# Each list is the base tables in the existing DDL; TPCH's revenue0 is a view.
+# The benchmark name is validated above.
+# shellcheck disable=SC2249
+case "${benchmark}" in
+ ssb)
+ tables=(customer part supplier dates lineorder)
+ suites=(ssb ssb-flat)
+ query_ids=(1.1 1.2 1.3 2.1 2.2 2.3 3.1 3.2 3.3 3.4 4.1 4.2 4.3)
+ generator_properties="'trino.connector.name'='ssb'"
+ ;;
+ tpch)
+ tables=(region nation supplier customer part partsupp orders lineitem)
+ single_split_tables=(region nation)
+ suites=(tpch)
+ mapfile -t query_ids < <(seq 1 22)
+ # Match the existing Doris DDL, including exact two-decimal monetary
values.
+ generator_properties="'trino.connector.name'='tpch',
+ 'trino.tpch.column-naming'='STANDARD',
+ 'trino.tpch.double-type-mapping'='DECIMAL'"
+ split_property='trino.tpch.splits-per-node'
+ ;;
+ tpcds)
+ tables=(call_center catalog_page catalog_returns catalog_sales
customer_address
+ customer_demographics customer date_dim household_demographics
income_band
+ inventory item promotion reason ship_mode store_returns
store_sales store
+ time_dim warehouse web_page web_returns web_sales web_site)
+ # Trino 435 uses tpcds 1.4: Parallel.splitWork emits rows only in
chunk 1
+ # when Scaling.getRowCount is below one million. These tables stay
below
+ # that threshold at every supported scale; SF1 has additional such
tables.
+ single_split_tables=(call_center catalog_page date_dim
household_demographics
+ income_band item promotion reason ship_mode store time_dim
warehouse web_page web_site)
+ if [[ ${scale} -eq 1 ]]; then
+ single_split_tables+=(catalog_returns catalog_sales
customer_address customer
+ store_returns store_sales web_returns web_sales)
+ fi
+ suites=(tpcds)
+ query_ids=()
+ for ((query = 1; query <= 99; query++)); do
+ query_ids+=("${query}")
+ # The existing suite splits these four templates into two
statements.
+ case "${query}" in
+ 14 | 23 | 24 | 39) query_ids+=("${query}_1") ;;
+ *) ;;
+ esac
+ done
+ generator_properties="'trino.connector.name'='tpcds'"
+ split_property='trino.tpcds.split-count'
+ ;;
+esac
+
+query_path() {
+ # The benchmark name is validated above.
+ # shellcheck disable=SC2249
+ case "${benchmark}" in
+ ssb) printf '%s/%s-queries/q%s.sql' "${SUITE_ROOT}" "$1" "$2" ;;
+ tpch) printf '%s/queries/q%s.sql' "${SUITE_ROOT}" "$2" ;;
+ tpcds) printf '%s/queries/sf%s/query%s.sql' "${SUITE_ROOT}" "${scale}"
"$2" ;;
+ esac
+}
+
+# Fail before creating a database if a requested suite is incomplete.
+for suite in "${suites[@]}"; do
+ for query in "${query_ids[@]}"; do
+ test -r "$(query_path "${suite}" "${query}")"
+ done
+done
+test -r "${ddl}"
+# shellcheck disable=SC1090
+source "${config}"
+database=${database:-${DB}}
+if [[ ! ${database} =~ ^[a-zA-Z_][a-zA-Z0-9_]*$ ]]; then
+ echo 'The database name must contain only letters, digits and
underscores.' >&2
+ exit 1
+fi
+export MYSQL_PWD=${PASSWORD:-}
+mysql_args=(--protocol=tcp --host="${FE_HOST}" --port="${FE_QUERY_PORT}"
--user="${USER}"
+ --batch --raw --skip-column-names --comments --skip-force)
+command -v mysql >/dev/null
+mkdir -p "$(dirname "${result_dir}")"
+# Refuse to overwrite an earlier run's measurements or query results.
+mkdir "${result_dir}"
+result_dir=$(cd "${result_dir}" && pwd)
+# Keep the log location visible even when the failing command redirects stderr.
+exec {error_fd}>&2
+trap 'echo "${benchmark_name} failed at line ${LINENO}. Logs: ${result_dir}.
Prepared tables are preserved." >&"${error_fd}"' ERR
+
+run_sql() {
+ local sql=$1
+ shift
+ mysql "${mysql_args[@]}" "$@" --execute "${sql}"
+}
+
+create_generator_catalog() {
+ local properties=${generator_properties}
+ if [[ ${benchmark} != ssb ]]; then
+ properties+=", '${split_property}'='$2'"
+ fi
+ run_sql "CREATE CATALOG \`$1\` PROPERTIES (
+ 'type'='trino-connector', ${properties});"
>>"${result_dir}/prepare.log" 2>&1
+}
+
+insert_sql() {
+ # A successful INSERT can otherwise return COMMITTED before its rows are
visible.
+ # Preserve the diagnostic and abort rather than benchmark incomplete data
or retry a write.
+ local output insert_status status=0
+ output=$(
+ set -e
+ run_sql "SET enable_insert_strict=true; SET insert_max_filter_ratio=0;
+ SET insert_timeout=14400; SET insert_visible_timeout_ms=600000; $1"
-vvv 2>&1
+ ) || status=$?
+ printf '%s\n' "${output}" >>"${result_dir}/prepare.log"
+ if [[ ${status} -ne 0 ]]; then
+ printf '%s\n' "${output}" >&2
+ return "${status}"
+ fi
+ # mysql -vvv also echoes the SQL. Read the status field only from returned
+ # metadata records, so SQL identifiers and label values cannot match it.
+ insert_status=$(awk -F"'" '
+ /^[[:space:]]*\{/ {
+ for (i = 2; i + 2 <= NF; i += 2) {
+ if ($i == "status" && $(i + 1) ~
/^[[:space:]]*:[[:space:]]*$/) {
+ print $(i + 2)
+ break
+ }
+ }
+ }' <<<"${output}")
+ if [[ ${insert_status} != VISIBLE ]]; then
+ echo "INSERT did not report VISIBLE (status:
${insert_status:-missing}); inspect prepare.log before continuing." >&2
+ return 1
+ fi
+}
+
+echo "${benchmark_name} SF${scale}: ${FE_HOST}:${FE_QUERY_PORT}/${database};
results: ${result_dir}"
+if [[ ${queries_only} -eq 0 ]]; then
+ if [[ ${benchmark} != ssb ]]; then
+ echo "Generator splits for parallel tables: ${generator_splits}; small
tables: 1" |
+ tee -a "${result_dir}/prepare.log"
+ fi
+ echo "Creating database and Trino ${benchmark_name} catalog"
+ # CREATE DATABASE is deliberately not IF NOT EXISTS: a retry must never
append
+ # the same generated data to these DUPLICATE KEY tables, even after a
partial run.
+ run_sql "CREATE DATABASE \`${database}\`;" >>"${result_dir}/prepare.log"
2>&1
+ generator_catalog="${benchmark}_gen_${database}"
+ create_generator_catalog "${generator_catalog}" "${generator_splits}"
+ single_split_catalog=${generator_catalog}
+ if [[ ${benchmark} != ssb && ${generator_splits} -gt 1 ]]; then
+ # Connector split properties belong to a catalog, not an individual
scan.
+ # Reuse one single-split catalog for all small tables.
+ single_split_catalog="${benchmark}_one_${database}"
+ create_generator_catalog "${single_split_catalog}" 1
+ fi
+ mysql_args+=(--database="${database}")
+ mysql "${mysql_args[@]}" <"${ddl}" \
+ >>"${result_dir}/prepare.log" 2>&1
+
+ for table in "${tables[@]}"; do
+ source_catalog=${generator_catalog}
+ if [[ ${benchmark} != ssb ]]; then
+ table_splits=${generator_splits}
+ if [[ " ${single_split_tables[*]} " == *" ${table} "* ]]; then
+ source_catalog=${single_split_catalog}
+ table_splits=1
+ fi
+ echo "Generating and importing ${table} (${table_splits} generator
splits)" |
+ tee -a "${result_dir}/prepare.log"
+ else
+ echo "Generating and importing ${table}"
+ fi
+ # Use target column names explicitly: source column order must not
affect the import.
+ # The backticks in sed quote SQL identifiers, not shell commands.
+ # shellcheck disable=SC2016
+ columns=$(
+ set -e
+ run_sql "DESC \`${table}\`;" 2>>"${result_dir}/prepare.log" |
+ cut -f1 | sed 's/.*/`&`/' | paste -sd,
+ )
+ source_columns=${columns}
+ if [[ ${benchmark} == tpcds && ${table} == promotion ]]; then
+ # Trino 435's generator misspells this column. Keep Doris's
existing
+ # schema and map the source name without relying on source column
order.
+ source_columns=${columns//p_response_target/p_response_targe}
+ fi
+ insert_sql "INSERT INTO \`${table}\` (${columns})
+ SELECT ${source_columns} FROM
\`${source_catalog}\`.\`sf${scale}\`.\`${table}\`;"
+ done
+
+ if [[ ${benchmark} == ssb && ${mode} != ssb ]]; then
+ echo 'Building lineorder_flat'
+ mysql "${mysql_args[@]}"
<"${SUITE_ROOT}/ddl/create-ssb-flat-tables-sf${scale}.sql" \
+ >>"${result_dir}/prepare.log" 2>&1
+ # Keep the existing loader's year-sized joins to bound each INSERT's
memory use.
+ for year in 1992 1993 1994 1995 1996 1997 1998; do
+ flat_sql=$(sed "s/@YEAR@/${year}/g; s/@NEXT_YEAR@/$((year + 1))/g"
\
+ "${SUITE_ROOT}/ddl/insert-ssb-flat.sql")
+ insert_sql "${flat_sql}"
+ done
+ fi
+ echo 'Collecting statistics'
+ run_sql "ANALYZE DATABASE \`${database}\` WITH FULL WITH SYNC;"
>>"${result_dir}/prepare.log" 2>&1
+else
+ mysql_args+=(--database="${database}")
+fi
+
+# Repeated runs should execute the query while still benefiting from warm data
caches.
+mysql_args+=(--init-command='SET enable_sql_cache=false,
enable_query_cache=false')
+run_sql 'SELECT VERSION(); SHOW VARIABLES; SHOW TABLE STATUS;'
>"${result_dir}/environment.txt"
+printf 'suite,query,cold_ms,hot1_ms,hot2_ms,best_hot_ms\n'
>"${result_dir}/result.csv"
+for suite in "${suites[@]}"; do
+ if [[ (${mode} == ssb && ${suite} == ssb-flat) || (${mode} == flat &&
${suite} == ssb) ]]; then
+ continue
+ fi
+ mkdir "${result_dir}/${suite}"
+ cold_total=0
+ hot_total=0
+ for query in "${query_ids[@]}"; do
+ times=()
+ for run in cold hot1 hot2; do
+ start=$(date +%s%3N)
+ mysql "${mysql_args[@]}" <"$(query_path "${suite}" "${query}")" \
+ >"${result_dir}/${suite}/q${query}.${run}.out" \
+ 2>"${result_dir}/${suite}/q${query}.${run}.err"
+ end=$(date +%s%3N)
+ times+=("$((end - start))")
+ done
+ best=$((times[1] < times[2] ? times[1] : times[2]))
+ printf '%s,q%s,%s,%s,%s,%s\n' "${suite}" "${query}" "${times[@]}"
"${best}" |
+ tee -a "${result_dir}/result.csv"
+ cold_total=$((cold_total + times[0]))
+ hot_total=$((hot_total + best))
+ done
+ echo "${suite}: total cold ${cold_total} ms, total best hot ${hot_total}
ms"
+done
+echo "${benchmark_name} completed. Results: ${result_dir}/result.csv"
diff --git a/tools/benchmark/tests/test_run_benchmark.py
b/tools/benchmark/tests/test_run_benchmark.py
new file mode 100644
index 00000000000..a6a90a2178f
--- /dev/null
+++ b/tools/benchmark/tests/test_run_benchmark.py
@@ -0,0 +1,380 @@
+#!/usr/bin/env python3
+# 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.
+
+"""Exercise orchestration failures without requiring a running Doris
cluster."""
+
+import csv
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+import unittest
+
+
+TOOLS_ROOT = Path(__file__).resolve().parents[2]
+MYSQL = r'''#!/usr/bin/env python3
+import json, os, sys
+args = sys.argv[1:]
+sql = args[args.index('--execute') + 1] if '--execute' in args else
sys.stdin.read()
+sql = '\n'.join(line for line in sql.splitlines() if not
line.lstrip().startswith('--')).strip()
+with open(os.environ['BENCHMARK_TEST_CALLS'], 'a') as log:
+ log.write(json.dumps({'sql': sql, 'args': args}) + '\n')
+scenario = os.environ.get('BENCHMARK_TEST_SCENARIO', '')
+if sql.startswith('CREATE DATABASE') and scenario == 'existing_database':
+ sys.exit('Database already exists')
+if sql.startswith('CREATE CATALOG'):
+ if scenario == 'catalog_error' or (scenario == 'single_catalog_error' and
'_one_' in sql):
+ sys.exit('Generator plugin is unavailable')
+if 'create table' in sql.lower() and scenario == 'ddl_error':
+ sys.exit('Table creation failed')
+if sql.startswith('ANALYZE') and scenario == 'analyze_error':
+ sys.exit('Statistics collection failed')
+if sql.startswith('DESC'):
+ if scenario == 'desc_error':
+ sys.exit('Cannot describe target table')
+ if '`promotion`' in sql:
+ print('p_promo_sk\tINT\np_response_target\tINT')
+ else:
+ print('lo_orderkey\tBIGINT\nlo_linenumber\tINT')
+elif sql.startswith('CREATE TABLE') and 'lineorder_flat' in sql and scenario
== 'flat_ddl_error':
+ sys.exit('Wide table creation failed')
+elif 'INSERT INTO' in sql:
+ if '-vvv' in args:
+ print('--------------\n' + sql + '\n--------------')
+ if scenario == 'insert_error' or (scenario == 'flat_insert_error' and
'lineorder_flat' in sql):
+ sys.exit('Generator exited with 23')
+ print('Query OK, 10 rows affected')
+ if scenario != 'missing_status':
+ status = ('COMMITTED' if scenario == 'unpublished' or
+ (scenario == 'flat_unpublished' and 'lineorder_flat' in sql)
else
+ 'PREPARE' if scenario == 'unexpected_status' else 'VISIBLE')
+ label = 'COMMITTED_label' if scenario == 'committed_label' else
'insert_label'
+ print("{'label':'%s', 'status':'%s', 'txnId':'123'}" % (label, status))
+elif sql.lower().startswith(('select', 'with')) and 'VERSION()' not in sql:
+ if scenario == 'query_error':
+ sys.exit('Query execution failed')
+ print('42')
+'''
+
+
+class RunBenchmarkTest(unittest.TestCase):
+ def setUp(self):
+ self.temporary = tempfile.TemporaryDirectory(prefix="benchmark run ")
+ self.addCleanup(self.temporary.cleanup)
+ self.root = Path(self.temporary.name)
+ (self.root / "mysql").write_text(MYSQL)
+ (self.root / "mysql").chmod(0o755)
+ self.config = self.root / "cluster.conf"
+
self.config.write_text("FE_HOST=127.0.0.1\nFE_QUERY_PORT=9030\nUSER=root\nPASSWORD=''\nDB=benchmark_test\n")
+ self.calls = self.root / "calls.jsonl"
+ self.results = self.root / "results"
+
+ def run_benchmark(self, scenario="", *options, benchmark="ssb"):
+ return subprocess.run(
+ ["bash", str(TOOLS_ROOT /
f"{benchmark}-tools/bin/run-{benchmark}.sh"), "-c", str(self.config),
+ "--result-dir", str(self.results), *options],
+ env=dict(os.environ, PATH=f"{self.root}:{os.environ['PATH']}",
+ BENCHMARK_TEST_CALLS=str(self.calls),
BENCHMARK_TEST_SCENARIO=scenario),
+ cwd=self.root, capture_output=True, text=True, timeout=30,
+ )
+
+ def statements(self):
+ return [json.loads(line)["sql"] for line in
self.calls.read_text().splitlines()]
+
+ def test_complete_workflow_and_distinct_results(self):
+ result = self.run_benchmark()
+ self.assertEqual(0, result.returncode, result.stdout + result.stderr)
+ calls = self.statements()
+ self.assertEqual(12, sum("INSERT INTO" in sql for sql in calls))
+ self.assertEqual(1, sum(sql.startswith("ANALYZE") for sql in calls))
+ with (self.results / "result.csv").open() as file:
+ rows = list(csv.DictReader(file))
+ self.assertEqual(26, len(rows))
+ self.assertEqual({"ssb", "ssb-flat"}, {row["suite"] for row in rows})
+ self.assertEqual(78, len(list(self.results.glob("*/*.out"))))
+
+ def test_existing_database_never_receives_inserts(self):
+ result = self.run_benchmark("existing_database")
+ self.assertNotEqual(0, result.returncode)
+ self.assertEqual(1, len(self.statements()))
+
+ def test_failed_or_unpublished_insert_stops_preparation(self):
+ for scenario in ("insert_error", "unpublished", "missing_status",
"unexpected_status"):
+ with self.subTest(scenario=scenario):
+ self.results = self.root / scenario
+ self.calls = self.root / f"{scenario}.jsonl"
+ result = self.run_benchmark(scenario)
+ self.assertNotEqual(0, result.returncode)
+ self.assertEqual(1, sum("INSERT INTO" in sql for sql in
self.statements()))
+ self.assertFalse((self.results / "result.csv").exists())
+ self.assertTrue((self.results / "prepare.log").read_text())
+
+ def test_visible_status_is_not_confused_with_sql_or_label(self):
+ for benchmark in ("ssb", "tpch", "tpcds"):
+ with self.subTest(benchmark=benchmark):
+ self.results = self.root / benchmark
+ self.calls = self.root / f"{benchmark}.jsonl"
+ result = self.run_benchmark("committed_label", "-d",
f"{benchmark}_COMMITTED",
+ benchmark=benchmark)
+ self.assertEqual(0, result.returncode, result.stdout +
result.stderr)
+ log = (self.results / "prepare.log").read_text()
+ self.assertIn(f"{benchmark}_gen_{benchmark}_COMMITTED", log)
+ self.assertIn("'label':'COMMITTED_label'", log)
+ self.assertTrue((self.results / "result.csv").exists())
+
+ def test_visible_in_sql_does_not_hide_an_unpublished_insert(self):
+ result = self.run_benchmark("unpublished", "-d", "tpch_VISIBLE",
benchmark="tpch")
+ self.assertNotEqual(0, result.returncode)
+ self.assertEqual(1, sum("INSERT INTO" in sql for sql in
self.statements()))
+ self.assertFalse((self.results / "result.csv").exists())
+
+ def test_generator_splits_scale_with_data_and_allow_override(self):
+ tpcds_small = {"call_center", "catalog_page", "date_dim",
"household_demographics",
+ "income_band", "item", "promotion", "reason",
"ship_mode", "store",
+ "time_dim", "warehouse", "web_page", "web_site"}
+ tpcds_sf1_small = {"catalog_returns", "catalog_sales",
"customer_address", "customer",
+ "store_returns", "store_sales", "web_returns",
"web_sales"}
+ for benchmark, property_name in (("tpch", "splits-per-node"),
("tpcds", "split-count")):
+ for scale, splits in ((1, 10), (100, 100), (1000, 1000), (10000,
10000)):
+ for override in (None, 7, 1):
+ with self.subTest(benchmark=benchmark, scale=scale,
override=override):
+ name = f"{benchmark}-{scale}-{override}"
+ self.results = self.root / name
+ self.calls = self.root / f"{name}.jsonl"
+ options = ("-s", str(scale))
+ if override is not None:
+ options += ("--splits", str(override))
+ # Complete every import, then stop before running
benchmark queries.
+ result = self.run_benchmark("analyze_error", *options,
benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ expected = override if override is not None else splits
+ calls = self.statements()
+ catalogs = {}
+ for sql in calls:
+ if sql.startswith("CREATE CATALOG"):
+ name = re.search(r"CREATE CATALOG `([^`]+)`",
sql).group(1)
+ count =
re.search(rf"'trino\.{benchmark}\.{property_name}'='(\d+)'", sql)
+ catalogs[name] = int(count.group(1))
+ if benchmark == "tpch":
+
self.assertIn("'trino.tpch.column-naming'='STANDARD'", sql)
+
self.assertIn("'trino.tpch.double-type-mapping'='DECIMAL'", sql)
+ self.assertEqual(1 if expected == 1 else 2,
len(catalogs))
+ small = {"region", "nation"} if benchmark == "tpch"
else tpcds_small
+ if benchmark == "tpcds" and scale == 1:
+ small = small | tpcds_sf1_small
+ inserts = [sql for sql in calls if "INSERT INTO" in
sql]
+ self.assertEqual(8 if benchmark == "tpch" else 24,
len(inserts))
+ log = (self.results / "prepare.log").read_text()
+ for sql in inserts:
+ table = re.search(r"INSERT INTO `([^`]+)`",
sql).group(1)
+ source = re.search(r"FROM
`([^`]+)`\.`sf(\d+)`\.`([^`]+)`", sql)
+ self.assertEqual((str(scale), table),
source.group(2, 3))
+ table_splits = 1 if table in small else expected
+ self.assertEqual(table_splits,
catalogs[source.group(1)], table)
+ self.assertIn(f"{table} ({table_splits} generator
splits)", log)
+ self.assertIn("Statistics collection failed", log)
+ self.assertFalse((self.results /
"result.csv").exists())
+ ddl = TOOLS_ROOT /
f"{benchmark}-tools/ddl/create-{benchmark}-tables-sf{scale}.sql"
+ expected_ddl = "\n".join(line for line in
ddl.read_text().splitlines()
+ if not
line.lstrip().startswith("--")).strip()
+ self.assertEqual(expected_ddl, next(sql for sql in
calls if "create table" in sql.lower()))
+
+ def test_preparation_errors_are_saved_in_log(self):
+ scenarios = {"existing_database": "Database already exists",
+ "catalog_error": "Generator plugin is unavailable",
+ "ddl_error": "Table creation failed",
+ "desc_error": "Cannot describe target table",
+ "analyze_error": "Statistics collection failed"}
+ for benchmark in ("ssb", "tpch", "tpcds"):
+ for scenario, error in scenarios.items():
+ with self.subTest(benchmark=benchmark, scenario=scenario):
+ self.results = self.root / f"{benchmark}-{scenario}"
+ self.calls = self.root / f"{benchmark}-{scenario}.jsonl"
+ result = self.run_benchmark(scenario, benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ self.assertIn(error, (self.results /
"prepare.log").read_text())
+ self.assertIn(str(self.results), result.stderr)
+ self.assertFalse((self.results / "result.csv").exists())
+ self.assertFalse(any("VERSION()" in sql for sql in
self.statements()))
+
+ def test_single_split_catalog_error_prevents_imports(self):
+ for benchmark in ("tpch", "tpcds"):
+ with self.subTest(benchmark=benchmark):
+ self.results = self.root / benchmark
+ self.calls = self.root / f"{benchmark}.jsonl"
+ result = self.run_benchmark("single_catalog_error",
benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ self.assertIn("Generator plugin is unavailable", (self.results
/ "prepare.log").read_text())
+ self.assertFalse(any("INSERT INTO" in sql for sql in
self.statements()))
+ self.assertFalse((self.results / "result.csv").exists())
+
+ def test_invalid_splits_are_rejected_before_connecting(self):
+ for benchmark in ("tpch", "tpcds"):
+ for splits in ("", "0", "-1", "1.5", "abc", "2147483648",
"9223372036854775808"):
+ with self.subTest(benchmark=benchmark, splits=splits):
+ result = self.run_benchmark("", "--splits", splits,
benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ self.assertFalse(self.calls.exists())
+ result = self.run_benchmark("", "--splits", "10", benchmark="ssb")
+ self.assertNotEqual(0, result.returncode)
+ self.assertFalse(self.calls.exists())
+
+ def test_query_failure_is_reported_and_stops_the_suite(self):
+ result = self.run_benchmark("query_error", "--queries-only")
+ self.assertNotEqual(0, result.returncode)
+ self.assertEqual(2, len(self.statements()))
+ self.assertIn("Query execution failed", (self.results /
"ssb/q1.1.cold.err").read_text())
+ self.assertNotIn("SSB completed", result.stdout)
+
+ def test_queries_only_does_not_modify_tables(self):
+ result = self.run_benchmark("", "--queries-only", "--mode", "flat")
+ self.assertEqual(0, result.returncode, result.stdout + result.stderr)
+ self.assertTrue(all(sql.startswith("SELECT") for sql in
self.statements()))
+ self.assertFalse((self.results / "ssb").exists())
+ self.assertEqual(39, len(list((self.results /
"ssb-flat").glob("*.out"))))
+
+ def test_invalid_scale_is_rejected_before_connecting(self):
+ result = self.run_benchmark("", "-s", "2")
+ self.assertNotEqual(0, result.returncode)
+ self.assertFalse(self.calls.exists())
+
+ def test_tpch_and_tpcds_complete_workflow(self):
+ for benchmark, table_count, query_count in (("tpch", 8, 22), ("tpcds",
24, 103)):
+ with self.subTest(benchmark=benchmark):
+ self.results = self.root / benchmark
+ self.calls = self.root / f"{benchmark}.jsonl"
+ result = self.run_benchmark(benchmark=benchmark)
+ self.assertEqual(0, result.returncode, result.stdout +
result.stderr)
+ calls = self.statements()
+ inserts = [sql for sql in calls if "INSERT INTO" in sql]
+ self.assertEqual(table_count, len(inserts))
+ self.assertFalse(any("_flat" in sql for sql in calls))
+ self.assertFalse((self.results /
"flat-row-counts.tsv").exists())
+ self.assertTrue(all(f"`{benchmark}_gen_benchmark_test`.`sf1`."
in sql or
+ f"`{benchmark}_one_benchmark_test`.`sf1`."
in sql for sql in inserts))
+ # Target columns and source projections must use the same
explicit names.
+ ordinary_inserts = [sql for sql in inserts if "INSERT INTO
`promotion`" not in sql]
+ self.assertTrue(all("(`lo_orderkey`,`lo_linenumber`)" in sql
for sql in ordinary_inserts))
+ self.assertTrue(all("SELECT `lo_orderkey`,`lo_linenumber`" in
sql for sql in ordinary_inserts))
+ analyses = [sql for sql in calls if sql.startswith("ANALYZE")]
+ self.assertEqual(1, len(analyses))
+ self.assertIn("WITH FULL WITH SYNC", analyses[0])
+ self.assertGreater(max(i for i, sql in enumerate(calls) if
sql.startswith("ANALYZE")),
+ max(i for i, sql in enumerate(calls) if
"INSERT INTO" in sql))
+ with (self.results / "result.csv").open() as file:
+ rows = list(csv.DictReader(file))
+ self.assertEqual(query_count, len(rows))
+ self.assertEqual({benchmark}, {row["suite"] for row in rows})
+ for row in rows:
+ self.assertEqual(min(int(row["hot1_ms"]),
int(row["hot2_ms"])), int(row["best_hot_ms"]))
+ self.assertEqual(query_count * 3,
len(list(self.results.glob("*/*.out"))))
+ entries = [json.loads(line) for line in
self.calls.read_text().splitlines()]
+ query_entries = entries[-query_count * 3:]
+ self.assertTrue(all("--init-command=SET
enable_sql_cache=false, enable_query_cache=false"
+ in entry["args"] for entry in
query_entries))
+ if benchmark == "tpch":
+ self.assertIn("'trino.tpch.column-naming'='STANDARD'",
calls[1])
+
self.assertIn("'trino.tpch.double-type-mapping'='DECIMAL'", calls[1])
+ self.assertTrue(any("create view revenue0" in sql for sql
in calls))
+ else:
+ self.assertIn("'trino.tpcds.split-count'='10'", calls[1])
+ promotion = next(sql for sql in inserts if "INSERT INTO
`promotion`" in sql)
+ self.assertIn("(`p_promo_sk`,`p_response_target`)",
promotion)
+ self.assertIn("SELECT `p_promo_sk`,`p_response_targe`",
promotion)
+ for query in (14, 23, 24, 39):
+ self.assertIn(f"q{query}_1", [row["query"] for row in
rows])
+
+ def test_tpc_queries_only_uses_requested_scale(self):
+ for benchmark in ("tpch", "tpcds"):
+ with self.subTest(benchmark=benchmark):
+ self.results = self.root / benchmark
+ self.calls = self.root / f"{benchmark}.jsonl"
+ result = self.run_benchmark("", "--queries-only", "-s",
"10000", benchmark=benchmark)
+ self.assertEqual(0, result.returncode, result.stdout +
result.stderr)
+ calls = self.statements()
+ self.assertFalse(any("INSERT INTO" in sql or
sql.startswith(("CREATE", "ANALYZE"))
+ for sql in calls))
+ query_path = (TOOLS_ROOT /
"tpcds-tools/queries/sf10000/query1.sql" if benchmark == "tpcds"
+ else TOOLS_ROOT / "tpch-tools/queries/q1.sql")
+ expected = "\n".join(line for line in
query_path.read_text().splitlines()
+ if not
line.lstrip().startswith("--")).strip()
+ self.assertEqual(expected, calls[1])
+
+ def test_tpc_preparation_errors_abort_before_queries(self):
+ for benchmark in ("tpch", "tpcds"):
+ for scenario in ("existing_database", "insert_error",
"unpublished"):
+ with self.subTest(benchmark=benchmark, scenario=scenario):
+ self.results = self.root / f"{benchmark}-{scenario}"
+ self.calls = self.root / f"{benchmark}-{scenario}.jsonl"
+ result = self.run_benchmark(scenario, benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ self.assertFalse((self.results / "result.csv").exists())
+ self.assertFalse(any(sql.startswith("ANALYZE") for sql in
self.statements()))
+ expected_inserts = 0 if scenario == "existing_database"
else 1
+ self.assertEqual(expected_inserts, sum("INSERT INTO" in
sql for sql in self.statements()))
+
+ def test_ssb_flat_failures_abort_before_measuring_queries(self):
+ for scenario in ("flat_ddl_error", "flat_insert_error",
"flat_unpublished"):
+ with self.subTest(scenario=scenario):
+ self.results = self.root / scenario
+ self.calls = self.root / f"{scenario}.jsonl"
+ result = self.run_benchmark(scenario)
+ self.assertNotEqual(0, result.returncode)
+ self.assertFalse((self.results / "result.csv").exists())
+ self.assertNotIn("completed", result.stdout)
+ base_inserts = [sql for sql in self.statements()
+ if "INSERT INTO" in sql and "lineorder_flat"
not in sql]
+ self.assertEqual(5, len(base_inserts))
+ if scenario == "flat_ddl_error":
+ self.assertIn("Wide table creation failed", (self.results
/ "prepare.log").read_text())
+
+ def test_tpc_query_error_is_not_recorded_as_success(self):
+ for benchmark in ("tpch", "tpcds"):
+ with self.subTest(benchmark=benchmark):
+ self.results = self.root / benchmark
+ self.calls = self.root / f"{benchmark}.jsonl"
+ result = self.run_benchmark("query_error", "--queries-only",
benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ with (self.results / "result.csv").open() as file:
+ self.assertEqual([], list(csv.DictReader(file)))
+ self.assertIn("Query execution failed", (self.results /
benchmark / "q1.cold.err").read_text())
+ self.assertEqual(2, len(self.statements()))
+
+ def
test_tpc_rejects_ssb_modes_and_invalid_arguments_before_connecting(self):
+ for benchmark in ("tpch", "tpcds"):
+ for options in (("--mode", "flat"), ("--mode", "both"), ("-s",
"2"),
+ ("-d", "invalid-name"), ("unexpected",)):
+ with self.subTest(benchmark=benchmark, options=options):
+ result = self.run_benchmark("", *options,
benchmark=benchmark)
+ self.assertNotEqual(0, result.returncode)
+ self.assertFalse(self.calls.exists())
+
+ def test_existing_result_directory_is_not_overwritten(self):
+ self.results.mkdir()
+ marker = self.results / "result.csv"
+ marker.write_text("previous result")
+ result = self.run_benchmark("", "--queries-only", benchmark="tpch")
+ self.assertNotEqual(0, result.returncode)
+ self.assertEqual("previous result", marker.read_text())
+ self.assertFalse(self.calls.exists())
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tools/ssb-tools/.gitignore b/tools/ssb-tools/.gitignore
new file mode 100644
index 00000000000..fbca2253799
--- /dev/null
+++ b/tools/ssb-tools/.gitignore
@@ -0,0 +1 @@
+results/
diff --git a/tools/ssb-tools/README.md b/tools/ssb-tools/README.md
index 8f857057f99..f6131fe479b 100644
--- a/tools/ssb-tools/README.md
+++ b/tools/ssb-tools/README.md
@@ -17,7 +17,97 @@ specific language governing permissions and limitations
under the License.
-->
-# Usage
+# Run SSB with one command
+
+The Trino SSB connector generates rows while Doris reads them. It uses the same
+`ssb-dbgen` distribution as the existing tools, with stdout connected to the
scan
+through a bounded pipe. There are no intermediate `.tbl` files, Stream Load
+requests, or separate Trino servers.
+
+## Install the generator plugin once
+
+Build on Linux with the same CPU architecture as the Doris nodes. The build
needs
+JDK 17, Maven, GCC, Make, curl, patch, and sha256sum:
+
+```bash
+./trino-ssb/build.sh
+```
+
+Copy the resulting **entire directory** `trino-ssb/target/ssb` into the
configured
+Trino plugin directory on **every FE and BE**. For current Doris installations:
+
+```bash
+cp -r trino-ssb/target/ssb /path/to/fe/plugins/trino_plugins/
+cp -r trino-ssb/target/ssb /path/to/be/plugins/trino_plugins/
+```
+
+Start or restart these nodes after installing the plugin. It consists of a JAR,
+the native `dbgen` executable, and its distribution file. The executable must
+remain executable. No paths or executable commands are accepted as catalog
properties.
+
+## Create, import, and benchmark
+
+Set the Doris connection in `conf/doris-cluster.conf`, then run:
+
+```bash
+./bin/run-ssb.sh -s 1 -d ssb_sf1
+```
+
+This creates a new database and its `ssb_gen_ssb_sf1` catalog, creates the five
+SSB tables, and imports them using `INSERT INTO ... SELECT ...` from the
catalog's
+`sf1` schema. It then builds `lineorder_flat`, collects statistics
synchronously,
+and runs all 13 SSB queries and all 13 SSB Flat queries three times each.
+The supported scale factors are 1, 100, and 1000. Use `--mode ssb` or `--mode
flat`
+to select one benchmark; the default is `both`.
+
+Each run writes a new `results/<timestamp>-<pid>/` directory with:
+
+- `result.csv`: first-run, two repeated-run, and best repeated-run times in
milliseconds.
+- `ssb/` and `ssb-flat/`: each query's output and errors.
+- `prepare.log` and `environment.txt`: import diagnostics and Doris settings.
+
+Timing includes the mysql client connection and result transfer, as in the old
+scripts. The `cold` column means the first execution; the script does not flush
+data caches. SQL result caching and BE query caching are disabled for every
measured
+session. A query or import error stops the run with a nonzero exit status.
+
+Preparation requires a **new database**, so rerunning a failed or completed
load
+cannot append duplicate data. Tables and the catalog are preserved for
inspection.
+To rerun only the queries against completed data:
+
+```bash
+./bin/run-ssb.sh -d ssb_sf1 --queries-only
+```
+
+Use `-c /path/to/cluster.conf` for a separate connection configuration, and
+`--result-dir /path/to/new-directory` for a custom output directory. Existing
+result directories are never overwritten.
+
+## Generator consistency and tests
+
+The connector keeps the old tools' default of ten `lineorder` generator
partitions.
+Each dimension uses one generator. Changing the original dbgen partition count
+changes some random streams, so partition count is intentionally fixed here.
+The date generator runs in UTC on every BE. The native build enables dbgen's
+existing 64-bit code path for large scale factors and avoids an out-of-bounds
+usage-counter update for its SSB-only random streams. Neither change replaces
+the data-generation algorithm.
+
+The existing benchmark SQL is preserved. In particular, SSB Q1.3 filters the
+generator's `d_weeknuminyear`, whereas SSB Flat Q1.3 uses `weekofyear()`;
these week
+definitions differ, so their aggregate results are not expected to match.
+
+The plugin build runs its Java tests, including full SF1 checksums of all five
tables,
+handle serialization, projected columns, large keys, producer failures, and
early
+cursor closure. Run script orchestration tests with:
+
+```bash
+python3 -m unittest discover -s ../benchmark/tests -v
+```
+
+The shared runner also supports [TPCH and TPCDS](../benchmark/README.md).
+
+# Original file-based workflow
These scripts are used to make ssb and ssb flat test.
The ssb flat data comes from ssb tables by way of 'INSERT INTO ... SELECT
...'.
diff --git a/tools/ssb-tools/bin/run-ssb.sh b/tools/ssb-tools/bin/run-ssb.sh
new file mode 100755
index 00000000000..c292279d460
--- /dev/null
+++ b/tools/ssb-tools/bin/run-ssb.sh
@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+# 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.
+
+set -Eeuo pipefail
+SCRIPT_ROOT=$(cd "$(dirname "$0")" && pwd)
+exec bash "${SCRIPT_ROOT}/../../benchmark/run-benchmark.sh" ssb "$@"
diff --git a/tools/ssb-tools/ddl/insert-ssb-flat.sql
b/tools/ssb-tools/ddl/insert-ssb-flat.sql
new file mode 100644
index 00000000000..b8e958deb1b
--- /dev/null
+++ b/tools/ssb-tools/ddl/insert-ssb-flat.sql
@@ -0,0 +1,30 @@
+-- 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.
+
+INSERT INTO lineorder_flat
+SELECT
+ lo_orderdate, lo_orderkey, lo_linenumber, lo_custkey, lo_partkey,
lo_suppkey,
+ lo_orderpriority, lo_shippriority, lo_quantity, lo_extendedprice,
lo_ordtotalprice,
+ lo_discount, lo_revenue, lo_supplycost, lo_tax, lo_commitdate, lo_shipmode,
+ c_name, c_address, c_city, c_nation, c_region, c_phone, c_mktsegment,
+ s_name, s_address, s_city, s_nation, s_region, s_phone,
+ p_name, p_mfgr, p_category, p_brand, p_color, p_type, p_size, p_container
+FROM lineorder l
+INNER JOIN customer c ON c.c_custkey = l.lo_custkey
+INNER JOIN supplier s ON s.s_suppkey = l.lo_suppkey
+INNER JOIN part p ON p.p_partkey = l.lo_partkey
+WHERE lo_orderdate >= @YEAR@0101 AND lo_orderdate < @NEXT_YEAR@0101;
diff --git a/tools/ssb-tools/trino-ssb/.gitignore
b/tools/ssb-tools/trino-ssb/.gitignore
new file mode 100644
index 00000000000..2f7896d1d13
--- /dev/null
+++ b/tools/ssb-tools/trino-ssb/.gitignore
@@ -0,0 +1 @@
+target/
diff --git a/tools/ssb-tools/trino-ssb/build.sh
b/tools/ssb-tools/trino-ssb/build.sh
new file mode 100755
index 00000000000..f6ba55d9ea1
--- /dev/null
+++ b/tools/ssb-tools/trino-ssb/build.sh
@@ -0,0 +1,45 @@
+#!/usr/bin/env bash
+# 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.
+
+set -euo pipefail
+SSB_PLUGIN_ROOT=$(cd "$(dirname "$0")" && pwd)
+cd "${SSB_PLUGIN_ROOT}"
+mkdir -p target/native
+
+# The same generator distribution as ../bin/build-ssb-dbgen.sh. Pin its
contents
+# so a benchmark run cannot silently switch to a different data generator.
+archive=target/ssb-dbgen-linux.tar.gz
+if [[ ! -f "${archive}" ]]; then
+ curl --fail --location --retry 3 \
+
https://palo-cloud-repo-bd.bd.bcebos.com/baidu-doris-release/ssb-dbgen-linux.tar.gz
\
+ --output "${archive}"
+fi
+echo "9a28f5a24091fd55d7e2ea12850e0189d98d1e1c6361f84442353ebd7df8e0ef
${archive}" | sha256sum --check
+tar -xzf "${archive}" -C target/native --strip-components=1
+# The legacy source contains trailing whitespace; keep that out of the patch.
+patch -l -d target/native -p1 <dbgen-stream.patch
+parallel=$(($(nproc) / 4))
+parallel=$((parallel > 0 ? parallel : 1))
+make -C target/native -j"${parallel}" CC="${CC:-gcc} -std=gnu89 -fcommon" dbgen
+mvn -B -f pom.xml -T "${parallel}" package
+
+mkdir -p target/ssb
+cp target/trino-ssb.jar target/native/dbgen target/native/dists.dss target/ssb/
+cp target/native/README target/native/TPCH_README target/ssb/
+echo "SSB plugin: ${SSB_PLUGIN_ROOT}/target/ssb"
+echo 'Install this directory under plugins/trino_plugins/ on every FE and BE
before starting Doris.'
diff --git a/tools/ssb-tools/trino-ssb/dbgen-stream.patch
b/tools/ssb-tools/trino-ssb/dbgen-stream.patch
new file mode 100644
index 00000000000..30f5d895862
--- /dev/null
+++ b/tools/ssb-tools/trino-ssb/dbgen-stream.patch
@@ -0,0 +1,62 @@
+--- a/print.c
++++ b/print.c
+@@ -35,41 +35,2 @@
+- char upath[128];
+- FILE *res;
+-
+- if (updates)
+- {
+- if (update > 0) /* updates */
+- if ( insert_segments )
+- {
+- int this_segment;
+- if(strcmp(tdefs[table].name,"orders.tbl"))
+- this_segment=++insert_orders_segment;
+- else
+- this_segment=++insert_lineitem_segment;
+- sprintf(upath, "%s%c%s.u%d.%d",
+- env_config(PATH_TAG, PATH_DFLT),
+- PATH_SEP, tdefs[table].name,
update%10000,this_segment);
+- }
+- else
+- {
+- sprintf(upath, "%s%c%s.u%d",
+- env_config(PATH_TAG, PATH_DFLT),
+- PATH_SEP, tdefs[table].name, update);
+- }
+- else /* deletes */
+- if ( delete_segments )
+- {
+- ++delete_segment;
+- sprintf(upath, "%s%cdelete.u%d.%d",
+- env_config(PATH_TAG, PATH_DFLT),
PATH_SEP, -update%10000,
+- delete_segment);
+- }
+- else
+- {
+- sprintf(upath, "%s%cdelete.%d",
+- env_config(PATH_TAG, PATH_DFLT), PATH_SEP,
-update);
+- }
+- return(fopen(upath, "w"));
+- }
+- res = tbl_open(table, "w");
+- OPEN_CHECK(res, tdefs[table].name);
+- return(res);
++ /* Each connector scan generates exactly one table into its stdout pipe.
*/
++ return stdout;
+--- a/rnd.c
++++ b/rnd.c
+@@ -28 +28,4 @@
+- Seed[stream].usage += 1;
++ /* UnifInt maps SSB-only streams 97 and 98 to stream 0. They have no
++ * per-row usage counter; do not write beyond the Seed array. */
++ if (stream >= 0 && stream <= MAX_STREAM)
++ Seed[stream].usage += 1;
+--- a/config.h
++++ b/config.h
+@@ -100,0 +101,5 @@
++/* Use the existing 64-bit generator path, including large RNG skip counts. */
++#define SUPPORT_64BITS
++#define DSS_HUGE long long
++#define HUGE_COUNT 1
++#define HUGE_FORMAT "%lld"
diff --git a/tools/ssb-tools/trino-ssb/pom.xml
b/tools/ssb-tools/trino-ssb/pom.xml
new file mode 100644
index 00000000000..70003cc2f48
--- /dev/null
+++ b/tools/ssb-tools/trino-ssb/pom.xml
@@ -0,0 +1,104 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
https://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.doris</groupId>
+ <artifactId>trino-ssb</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ <properties>
+ <maven.compiler.release>17</maven.compiler.release>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <!-- Supplied by Doris's Trino plugin classloader; never bundle a
second SPI. -->
+ <dependency>
+ <groupId>io.trino</groupId>
+ <artifactId>trino-spi</artifactId>
+ <version>435</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>5.10.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.airlift</groupId>
+ <artifactId>json</artifactId>
+ <version>239</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>2.15.3</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ <build>
+ <finalName>trino-ssb</finalName>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>3.1.2</version>
+ <dependencies>
+ <dependency>
+ <groupId>com.puppycrawl.tools</groupId>
+ <artifactId>checkstyle</artifactId>
+ <version>9.3</version>
+ </dependency>
+ </dependencies>
+ <configuration>
+
<configLocation>${project.basedir}/../../../fe/check/checkstyle/checkstyle.xml</configLocation>
+
<propertyExpansion>config_loc=${project.basedir}/../../../fe/check/checkstyle</propertyExpansion>
+
<includeTestSourceDirectory>true</includeTestSourceDirectory>
+ <consoleOutput>true</consoleOutput>
+ </configuration>
+ <executions>
+ <execution>
+ <phase>validate</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.13.0</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>3.2.5</version>
+ <configuration>
+ <systemPropertyVariables>
+
<ssb.test.dbgen>${project.build.directory}/native/dbgen</ssb.test.dbgen>
+ </systemPropertyVariables>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbConnector.java
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbConnector.java
new file mode 100644
index 00000000000..1866c58672e
--- /dev/null
+++
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbConnector.java
@@ -0,0 +1,226 @@
+// 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.
+
+package org.apache.doris.tools.ssb;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.trino.spi.HostAddress;
+import io.trino.spi.SplitWeight;
+import io.trino.spi.connector.ColumnHandle;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.Connector;
+import io.trino.spi.connector.ConnectorMetadata;
+import io.trino.spi.connector.ConnectorRecordSetProvider;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorSplit;
+import io.trino.spi.connector.ConnectorSplitManager;
+import io.trino.spi.connector.ConnectorTableHandle;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.ConnectorTransactionHandle;
+import io.trino.spi.connector.FixedSplitSource;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.connector.SchemaTablePrefix;
+import io.trino.spi.transaction.IsolationLevel;
+
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+public final class SsbConnector implements Connector {
+ // Keep the original tool's ten dbgen partitions: changing -C changes some
random streams.
+ static final int LINEORDER_PARTS = 10;
+ private static final List<String> SCHEMAS = List.of("sf1", "sf100",
"sf1000");
+ private final Path dbgen;
+
+ SsbConnector(Path dbgen) {
+ this.dbgen = dbgen;
+ }
+
+ public enum Transaction implements ConnectorTransactionHandle {
+ INSTANCE
+ }
+
+ public record Table(@JsonProperty("schema") String schema,
@JsonProperty("name") String name)
+ implements ConnectorTableHandle {
+ @JsonCreator
+ public Table {
+ if (!SCHEMAS.contains(schema)) {
+ throw new IllegalArgumentException("Supported SSB schemas: " +
SCHEMAS);
+ }
+ SsbTable.fromName(name);
+ }
+ }
+
+ public record Column(@JsonProperty("index") int index) implements
ColumnHandle {
+ @JsonCreator
+ public Column {
+ }
+ }
+
+ public record Split(@JsonProperty("part") int part,
@JsonProperty("totalParts") int totalParts)
+ implements ConnectorSplit {
+ @JsonCreator
+ public Split {
+ if ((totalParts != 1 && totalParts != LINEORDER_PARTS) || part < 1
|| part > totalParts) {
+ throw new IllegalArgumentException("Invalid SSB partition " +
part + "/" + totalParts);
+ }
+ }
+
+ @Override
+ @JsonIgnore
+ public boolean isRemotelyAccessible() {
+ return true;
+ }
+
+ @Override
+ @JsonIgnore
+ public List<HostAddress> getAddresses() {
+ return List.of();
+ }
+
+ @Override
+ @JsonIgnore
+ public Object getInfo() {
+ return Map.of("part", part, "totalParts", totalParts);
+ }
+
+ @Override
+ @JsonIgnore
+ public long getRetainedSizeInBytes() {
+ return 24; // object header and two int fields
+ }
+
+ @Override
+ @JsonIgnore
+ public SplitWeight getSplitWeight() {
+ return SplitWeight.standard();
+ }
+ }
+
+ @Override
+ public ConnectorTransactionHandle beginTransaction(IsolationLevel
isolation, boolean readOnly, boolean autoCommit) {
+ return Transaction.INSTANCE;
+ }
+
+ @Override
+ public ConnectorMetadata getMetadata(ConnectorSession session,
ConnectorTransactionHandle transaction) {
+ return new Metadata();
+ }
+
+ @Override
+ public ConnectorSplitManager getSplitManager() {
+ return new ConnectorSplitManager() {
+ @Override
+ public io.trino.spi.connector.ConnectorSplitSource
getSplits(ConnectorTransactionHandle transaction,
+ ConnectorSession session, ConnectorTableHandle table,
io.trino.spi.connector.DynamicFilter filter,
+ io.trino.spi.connector.Constraint constraint) {
+ int count = SsbTable.fromName(((Table) table).name()) ==
SsbTable.LINEORDER ? LINEORDER_PARTS : 1;
+ List<Split> splits = new ArrayList<>();
+ for (int part = 1; part <= count; part++) {
+ splits.add(new Split(part, count));
+ }
+ return new FixedSplitSource(splits);
+ }
+ };
+ }
+
+ @Override
+ public ConnectorRecordSetProvider getRecordSetProvider() {
+ return new ConnectorRecordSetProvider() {
+ @Override
+ public io.trino.spi.connector.RecordSet
getRecordSet(ConnectorTransactionHandle transaction,
+ ConnectorSession session, ConnectorSplit split,
ConnectorTableHandle table,
+ List<? extends ColumnHandle> columns) {
+ return new SsbRecordSet(dbgen, (Table) table, (Split) split,
+ columns.stream().map(column -> ((Column)
column).index()).toList());
+ }
+ };
+ }
+
+ private static final class Metadata implements ConnectorMetadata {
+ @Override
+ public List<String> listSchemaNames(ConnectorSession session) {
+ return SCHEMAS;
+ }
+
+ @Override
+ public ConnectorTableHandle getTableHandle(ConnectorSession session,
SchemaTableName table) {
+ if (!SCHEMAS.contains(table.getSchemaName())) {
+ return null;
+ }
+ for (SsbTable ssbTable : SsbTable.values()) {
+ if (ssbTable.tableName().equals(table.getTableName())) {
+ return new Table(table.getSchemaName(),
table.getTableName());
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public List<SchemaTableName> listTables(ConnectorSession session,
Optional<String> schema) {
+ List<SchemaTableName> tables = new ArrayList<>();
+ for (String name : SCHEMAS) {
+ if (schema.isEmpty() || schema.get().equals(name)) {
+ for (SsbTable table : SsbTable.values()) {
+ tables.add(new SchemaTableName(name,
table.tableName()));
+ }
+ }
+ }
+ return tables;
+ }
+
+ @Override
+ public ConnectorTableMetadata getTableMetadata(ConnectorSession
session, ConnectorTableHandle handle) {
+ Table table = (Table) handle;
+ return new ConnectorTableMetadata(new
SchemaTableName(table.schema(), table.name()),
+ SsbTable.fromName(table.name()).columns);
+ }
+
+ @Override
+ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession
session, ConnectorTableHandle handle) {
+ List<ColumnMetadata> columns = getTableMetadata(session,
handle).getColumns();
+ Map<String, ColumnHandle> result = new LinkedHashMap<>();
+ for (int i = 0; i < columns.size(); i++) {
+ result.put(columns.get(i).getName(), new Column(i));
+ }
+ return result;
+ }
+
+ @Override
+ public ColumnMetadata getColumnMetadata(ConnectorSession session,
ConnectorTableHandle table,
+ ColumnHandle column) {
+ return getTableMetadata(session, table).getColumns().get(((Column)
column).index());
+ }
+
+ @Override
+ public Map<SchemaTableName, List<ColumnMetadata>>
listTableColumns(ConnectorSession session,
+ SchemaTablePrefix prefix) {
+ Map<SchemaTableName, List<ColumnMetadata>> result = new
LinkedHashMap<>();
+ for (SchemaTableName table : listTables(session,
prefix.getSchema())) {
+ if (prefix.matches(table)) {
+ result.put(table,
SsbTable.fromName(table.getTableName()).columns);
+ }
+ }
+ return result;
+ }
+ }
+}
diff --git
a/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbPlugin.java
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbPlugin.java
new file mode 100644
index 00000000000..4f806e0c685
--- /dev/null
+++
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbPlugin.java
@@ -0,0 +1,56 @@
+// 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.
+
+package org.apache.doris.tools.ssb;
+
+import io.trino.spi.Plugin;
+import io.trino.spi.connector.Connector;
+import io.trino.spi.connector.ConnectorContext;
+import io.trino.spi.connector.ConnectorFactory;
+
+import java.net.URISyntaxException;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+
+public final class SsbPlugin implements Plugin {
+ @Override
+ public Iterable<ConnectorFactory> getConnectorFactories() {
+ return List.of(new ConnectorFactory() {
+ @Override
+ public String getName() {
+ return "ssb";
+ }
+
+ @Override
+ public Connector create(String catalogName, Map<String, String>
config, ConnectorContext context) {
+ if (!config.isEmpty()) {
+ throw new IllegalArgumentException("SSB has no connector
properties: " + config.keySet());
+ }
+ try {
+ // Only the operator-installed binary next to this JAR is
executable. Catalog
+ // properties cannot select a command, a path, or
arbitrary dbgen arguments.
+ Path pluginJar =
Path.of(SsbPlugin.class.getProtectionDomain()
+ .getCodeSource().getLocation().toURI());
+ return new
SsbConnector(pluginJar.getParent().resolve("dbgen"));
+ } catch (URISyntaxException e) {
+ throw new IllegalStateException("Cannot locate the SSB
plugin directory", e);
+ }
+ }
+ });
+ }
+}
diff --git
a/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbRecordSet.java
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbRecordSet.java
new file mode 100644
index 00000000000..803eb29ef75
--- /dev/null
+++
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbRecordSet.java
@@ -0,0 +1,198 @@
+// 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.
+
+package org.apache.doris.tools.ssb;
+
+import io.airlift.slice.Slice;
+import io.airlift.slice.Slices;
+import io.trino.spi.connector.RecordCursor;
+import io.trino.spi.connector.RecordSet;
+import io.trino.spi.type.Type;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+final class SsbRecordSet implements RecordSet {
+ private final Path dbgen;
+ private final SsbConnector.Table table;
+ private final SsbConnector.Split split;
+ private final List<Integer> columns;
+ private final List<Type> types;
+
+ SsbRecordSet(Path dbgen, SsbConnector.Table table, SsbConnector.Split
split, List<Integer> columns) {
+ this.dbgen = dbgen;
+ this.table = table;
+ this.split = split;
+ this.columns = List.copyOf(columns);
+ this.types = columns.stream()
+ .map(index ->
SsbTable.fromName(table.name()).columns.get(index).getType()).toList();
+ }
+
+ @Override
+ public List<Type> getColumnTypes() {
+ return types;
+ }
+
+ @Override
+ public RecordCursor cursor() {
+ return new Cursor();
+ }
+
+ private final class Cursor implements RecordCursor {
+ private final Process process;
+ private final BufferedReader reader;
+ private final int fieldCount =
SsbTable.fromName(table.name()).columns.size();
+ private long completedBytes;
+ private long readTimeNanos;
+ private String[] fields;
+ private boolean closed;
+
+ Cursor() {
+ ProcessBuilder builder = new ProcessBuilder(dbgen.toString(),
"-q", "-s", table.schema().substring(2),
+ "-T", SsbTable.fromName(table.name()).dbgenOption,
+ "-b", dbgen.getParent().resolve("dists.dss").toString(),
+ "-C", Integer.toString(split.totalParts()), "-S",
Integer.toString(split.part()));
+ // mk_date uses localtime. All BEs must generate the same date
dimension.
+ builder.environment().put("TZ", "UTC");
+ // Drain diagnostics without an extra thread or an unbounded
stderr buffer.
+ builder.redirectError(ProcessBuilder.Redirect.INHERIT);
+ try {
+ process = builder.start();
+ reader = new BufferedReader(new
InputStreamReader(process.getInputStream(), StandardCharsets.US_ASCII),
+ 64 * 1024);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Cannot start the
operator-installed SSB generator: " + dbgen, e);
+ }
+ }
+
+ @Override
+ public boolean advanceNextPosition() {
+ if (closed) {
+ return false;
+ }
+ long start = System.nanoTime();
+ try {
+ String line = reader.readLine();
+ if (line == null) {
+ int status = process.waitFor();
+ close();
+ if (status != 0) {
+ throw new IllegalStateException("SSB generator exited
with " + status
+ + " for " + table + ", partition " + split);
+ }
+ return false;
+ }
+ fields = line.split("\\|", -1);
+ if (fields.length != fieldCount + 1 ||
!fields[fieldCount].isEmpty()) {
+ throw new IllegalStateException("Invalid SSB row for " +
table + ", partition " + split);
+ }
+ completedBytes += line.length() + 1;
+ return true;
+ } catch (IOException e) {
+ close();
+ throw new UncheckedIOException("Cannot read SSB data for " +
table + ", partition " + split, e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ close();
+ throw new IllegalStateException("SSB generation interrupted",
e);
+ } catch (RuntimeException e) {
+ close();
+ throw e;
+ } finally {
+ readTimeNanos += System.nanoTime() - start;
+ }
+ }
+
+ @Override
+ public long getCompletedBytes() {
+ return completedBytes;
+ }
+
+ @Override
+ public long getReadTimeNanos() {
+ return readTimeNanos;
+ }
+
+ @Override
+ public Type getType(int field) {
+ return types.get(field);
+ }
+
+ @Override
+ public long getLong(int field) {
+ return Long.parseLong(fields[columns.get(field)]);
+ }
+
+ @Override
+ public Slice getSlice(int field) {
+ return Slices.utf8Slice(fields[columns.get(field)]);
+ }
+
+ @Override
+ public boolean isNull(int field) {
+ columns.get(field);
+ return false;
+ }
+
+ @Override
+ public boolean getBoolean(int field) {
+ throw new UnsupportedOperationException("SSB has no Boolean
columns");
+ }
+
+ @Override
+ public double getDouble(int field) {
+ throw new UnsupportedOperationException("SSB has no floating point
columns");
+ }
+
+ @Override
+ public Object getObject(int field) {
+ throw new UnsupportedOperationException("SSB has no object
columns");
+ }
+
+ @Override
+ public void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ // LIMIT, cancellation and scanner errors must also stop the
producer. Killing before
+ // closing the reader unblocks a producer waiting for space in the
pipe.
+ process.destroy();
+ try {
+ if (!process.waitFor(2, TimeUnit.SECONDS)) {
+ process.destroyForcibly();
+ process.waitFor();
+ }
+ } catch (InterruptedException e) {
+ process.destroyForcibly();
+ Thread.currentThread().interrupt();
+ } finally {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new UncheckedIOException("Cannot close the SSB
generator pipe", e);
+ }
+ }
+ }
+ }
+}
diff --git
a/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbTable.java
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbTable.java
new file mode 100644
index 00000000000..85174519631
--- /dev/null
+++
b/tools/ssb-tools/trino-ssb/src/main/java/org/apache/doris/tools/ssb/SsbTable.java
@@ -0,0 +1,60 @@
+// 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.
+
+package org.apache.doris.tools.ssb;
+
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.type.BigintType;
+import io.trino.spi.type.VarcharType;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Locale;
+
+/** Column order is the ssb-dbgen output order, also used by the existing SSB
DDL. */
+enum SsbTable {
+ CUSTOMER("c", "c_custkey", "c_name:s", "c_address:s", "c_city:s",
"c_nation:s", "c_region:s",
+ "c_phone:s", "c_mktsegment:s"),
+ PART("p", "p_partkey", "p_name:s", "p_mfgr:s", "p_category:s",
"p_brand:s", "p_color:s",
+ "p_type:s", "p_size", "p_container:s"),
+ SUPPLIER("s", "s_suppkey", "s_name:s", "s_address:s", "s_city:s",
"s_nation:s", "s_region:s", "s_phone:s"),
+ DATES("d", "d_datekey", "d_date:s", "d_dayofweek:s", "d_month:s",
"d_year", "d_yearmonthnum",
+ "d_yearmonth:s", "d_daynuminweek", "d_daynuminmonth",
"d_daynuminyear", "d_monthnuminyear",
+ "d_weeknuminyear", "d_sellingseason:s", "d_lastdayinweekfl",
"d_lastdayinmonthfl", "d_holidayfl",
+ "d_weekdayfl"),
+ LINEORDER("l", "lo_orderkey", "lo_linenumber", "lo_custkey", "lo_partkey",
"lo_suppkey", "lo_orderdate",
+ "lo_orderpriority:s", "lo_shippriority", "lo_quantity",
"lo_extendedprice", "lo_ordtotalprice",
+ "lo_discount", "lo_revenue", "lo_supplycost", "lo_tax",
"lo_commitdate", "lo_shipmode:s");
+
+ final String dbgenOption;
+ final List<ColumnMetadata> columns;
+
+ SsbTable(String dbgenOption, String... columns) {
+ this.dbgenOption = dbgenOption;
+ this.columns = Arrays.stream(columns).map(column ->
column.endsWith(":s")
+ ? new ColumnMetadata(column.substring(0, column.length() - 2),
VarcharType.VARCHAR)
+ : new ColumnMetadata(column, BigintType.BIGINT)).toList();
+ }
+
+ String tableName() {
+ return name().toLowerCase(Locale.ROOT);
+ }
+
+ static SsbTable fromName(String name) {
+ return valueOf(name.toUpperCase(Locale.ROOT));
+ }
+}
diff --git
a/tools/ssb-tools/trino-ssb/src/main/resources/META-INF/services/io.trino.spi.Plugin
b/tools/ssb-tools/trino-ssb/src/main/resources/META-INF/services/io.trino.spi.Plugin
new file mode 100644
index 00000000000..835218075ab
--- /dev/null
+++
b/tools/ssb-tools/trino-ssb/src/main/resources/META-INF/services/io.trino.spi.Plugin
@@ -0,0 +1,18 @@
+# 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.
+
+org.apache.doris.tools.ssb.SsbPlugin
diff --git
a/tools/ssb-tools/trino-ssb/src/test/java/org/apache/doris/tools/ssb/SsbConnectorTest.java
b/tools/ssb-tools/trino-ssb/src/test/java/org/apache/doris/tools/ssb/SsbConnectorTest.java
new file mode 100644
index 00000000000..b2d676d11aa
--- /dev/null
+++
b/tools/ssb-tools/trino-ssb/src/test/java/org/apache/doris/tools/ssb/SsbConnectorTest.java
@@ -0,0 +1,186 @@
+// 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.
+
+package org.apache.doris.tools.ssb;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.airlift.json.ObjectMapperProvider;
+import io.trino.spi.Plugin;
+import io.trino.spi.connector.ConnectorMetadata;
+import io.trino.spi.connector.Constraint;
+import io.trino.spi.connector.DynamicFilter;
+import io.trino.spi.connector.RecordCursor;
+import io.trino.spi.connector.RecordSet;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.type.BigintType;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+class SsbConnectorTest {
+ private final Path dbgen = Path.of(System.getProperty("ssb.test.dbgen"));
+
+ @TempDir
+ Path temporary;
+
+ @Test
+ void metadataAndPluginDiscovery() throws Exception {
+ Plugin plugin = ServiceLoader.load(Plugin.class).stream()
+ .map(ServiceLoader.Provider::get).filter(p -> p instanceof
SsbPlugin).findFirst().orElseThrow();
+ var factory = plugin.getConnectorFactories().iterator().next();
+ Assertions.assertEquals("ssb", factory.getName());
+ Assertions.assertThrows(IllegalArgumentException.class, () ->
factory.create("ssb", Map.of("command", "anything"), null));
+
+ var connector = new SsbConnector(dbgen);
+ ConnectorMetadata metadata = connector.getMetadata(null,
SsbConnector.Transaction.INSTANCE);
+ Assertions.assertEquals(List.of("sf1", "sf100", "sf1000"),
metadata.listSchemaNames(null));
+ Assertions.assertEquals(15, metadata.listTables(null,
Optional.empty()).size());
+ Assertions.assertTrue(metadata.listTables(null,
Optional.of("sf2")).isEmpty());
+ Assertions.assertNull(metadata.getTableHandle(null, new
SchemaTableName("sf2", "customer")));
+ Assertions.assertNull(metadata.getTableHandle(null, new
SchemaTableName("sf1", "unknown")));
+ var handle = metadata.getTableHandle(null, new SchemaTableName("sf1",
"lineorder"));
+ Assertions.assertNotNull(handle);
+ Assertions.assertEquals(17, metadata.getColumnHandles(null,
handle).size());
+ try (var splits =
connector.getSplitManager().getSplits(SsbConnector.Transaction.INSTANCE, null,
handle,
+ DynamicFilter.EMPTY, Constraint.alwaysTrue())) {
+ Assertions.assertEquals(10,
splits.getNextBatch(100).get().getSplits().size());
+ Assertions.assertTrue(splits.isFinished());
+ }
+ }
+
+ @Test
+ void handlesRoundTripBetweenFrontendAndBackend() throws Exception {
+ // Trino disables Jackson's automatic field/getter/constructor
discovery.
+ ObjectMapper mapper = new ObjectMapperProvider().get();
+ for (Object handle : List.of(new SsbConnector.Table("sf1000",
"lineorder"),
+ new SsbConnector.Column(16), new SsbConnector.Split(10, 10),
SsbConnector.Transaction.INSTANCE)) {
+ Assertions.assertEquals(handle,
mapper.readValue(mapper.writeValueAsBytes(handle), handle.getClass()));
+ }
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
SsbConnector.Table("sf1; anything", "lineorder"));
+ Assertions.assertThrows(IllegalArgumentException.class, () -> new
SsbConnector.Split(0, 10));
+ }
+
+ @Test
+ void projectionPreservesColumnOrderAndTypes() {
+ RecordSet records = new SsbRecordSet(dbgen, new
SsbConnector.Table("sf1", "customer"),
+ new SsbConnector.Split(1, 1), List.of(7, 0, 1));
+ try (RecordCursor cursor = records.cursor()) {
+ Assertions.assertTrue(cursor.advanceNextPosition());
+ Assertions.assertEquals("BUILDING",
cursor.getSlice(0).toStringUtf8());
+ Assertions.assertEquals(1, cursor.getLong(1));
+ Assertions.assertEquals("Customer#000000001",
cursor.getSlice(2).toStringUtf8());
+ Assertions.assertFalse(cursor.isNull(0));
+ Assertions.assertTrue(cursor.getCompletedBytes() > 0);
+ Assertions.assertTrue(cursor.getReadTimeNanos() > 0);
+ }
+ }
+
+ @Test
+ void dimensionsMatchTheExistingGenerator() throws Exception {
+ // SHA-256 of the existing tool's individual-table SF1 output
(TZ=UTC), including
+ // its trailing pipe and newline. Check all rows, not just a generated
sample.
+ verifyTable(SsbTable.CUSTOMER, 30000,
"e84ecfcde0f73bbc48f710ffe1c6d340321e875171013b272cbccee45f1d3cbc");
+ verifyTable(SsbTable.PART, 200000,
"9fb468e8a3e6ca9f5b6f841cf5a81276315fd241fbfdb35c8839328729bb816b");
+ verifyTable(SsbTable.SUPPLIER, 2000,
"90b636621dc449bedce76f347af5802701db6f6334d209f002d73f4044cf9404");
+ verifyTable(SsbTable.DATES, 2556,
"7057acc5ebc0b92225a4937488028767f46e9f8719126a8e29d4bf164e5461d1");
+ }
+
+ private void verifyTable(SsbTable table, long expectedRows, String
expectedDigest) throws Exception {
+ verifyTable(table, 1, expectedRows, expectedDigest);
+ }
+
+ @Test
+ void lineorderMatchesTheExistingTenPartitionLoad() throws Exception {
+ verifyTable(SsbTable.LINEORDER, 10, 6001215,
+
"67243ab009a55fe0a848a15404233789b0b48ff11ae913f85e22957a7751c7e8");
+ }
+
+ private void verifyTable(SsbTable table, int partitions, long
expectedRows, String expectedDigest) throws Exception {
+ MessageDigest digest = MessageDigest.getInstance("SHA-256");
+ long rows = 0;
+ for (int part = 1; part <= partitions; part++) {
+ RecordSet records = new SsbRecordSet(dbgen, new
SsbConnector.Table("sf1", table.tableName()),
+ new SsbConnector.Split(part, partitions),
IntStream.range(0, table.columns.size()).boxed().toList());
+ try (RecordCursor cursor = records.cursor()) {
+ while (cursor.advanceNextPosition()) {
+ StringBuilder row = new StringBuilder();
+ for (int i = 0; i < table.columns.size(); i++) {
+ row.append(cursor.getType(i).equals(BigintType.BIGINT)
+ ? Long.toString(cursor.getLong(i)) :
cursor.getSlice(i).toStringUtf8()).append('|');
+ }
+
digest.update(row.append('\n').toString().getBytes(StandardCharsets.US_ASCII));
+ rows++;
+ }
+ }
+ }
+ Assertions.assertEquals(expectedRows, rows, table.tableName());
+ Assertions.assertEquals(expectedDigest,
HexFormat.of().formatHex(digest.digest()), table.tableName());
+ }
+
+ @Test
+ void lineorderPartitionsAndLargeKeys() {
+ // The last partition starts after 90% of the orders. Exercise SF1000
without
+ // materializing billions of rows, including order keys beyond signed
INT.
+ try (RecordCursor cursor = new SsbRecordSet(dbgen, new
SsbConnector.Table("sf1000", "lineorder"),
+ new SsbConnector.Split(10, 10), List.of(0, 1, 2, 3,
4)).cursor()) {
+ Assertions.assertTrue(cursor.advanceNextPosition());
+ Assertions.assertEquals(5400000001L, cursor.getLong(0));
+ Assertions.assertEquals(1, cursor.getLong(1));
+ Assertions.assertTrue(cursor.getLong(2) <= 30000000);
+ Assertions.assertTrue(cursor.getLong(4) <= 2000000);
+ }
+ }
+
+ @Test
+ void nonzeroExitIsNotTreatedAsEndOfData() throws Exception {
+ Path failingGenerator = temporary.resolve("dbgen");
+ Files.writeString(failingGenerator, "#!/bin/sh\nexit 23\n");
+ Assertions.assertTrue(failingGenerator.toFile().setExecutable(true));
+ try (RecordCursor cursor = new SsbRecordSet(failingGenerator, new
SsbConnector.Table("sf1", "customer"),
+ new SsbConnector.Split(1, 1), List.of(0)).cursor()) {
+
Assertions.assertTrue(Assertions.assertThrows(IllegalStateException.class,
cursor::advanceNextPosition)
+ .getMessage().contains("23"));
+ }
+ }
+
+ @Test
+ void closingACursorStopsItsProducer() throws Exception {
+ var childPidsBefore =
ProcessHandle.current().children().map(ProcessHandle::pid).toList();
+ RecordCursor cursor = new SsbRecordSet(dbgen, new
SsbConnector.Table("sf1000", "lineorder"),
+ new SsbConnector.Split(1, 10), List.of(0)).cursor();
+ Assertions.assertTrue(cursor.advanceNextPosition());
+ var producer = ProcessHandle.current().children()
+ .filter(child ->
!childPidsBefore.contains(child.pid())).findFirst().orElseThrow();
+ cursor.close();
+ producer.onExit().get(5, TimeUnit.SECONDS);
+ Assertions.assertFalse(producer.isAlive());
+ cursor.close();
+ Assertions.assertFalse(cursor.advanceNextPosition());
+ }
+}
diff --git a/tools/tpcds-tools/.gitignore b/tools/tpcds-tools/.gitignore
new file mode 100644
index 00000000000..fbca2253799
--- /dev/null
+++ b/tools/tpcds-tools/.gitignore
@@ -0,0 +1 @@
+results/
diff --git a/tools/tpcds-tools/README.md b/tools/tpcds-tools/README.md
index 5c80a395bee..90b0f3b533d 100644
--- a/tools/tpcds-tools/README.md
+++ b/tools/tpcds-tools/README.md
@@ -17,6 +17,27 @@ specific language governing permissions and limitations
under the License.
-->
+# Run TPCDS with one command
+
+Install the Trino TPCDS generator on every FE and BE using the
+[shared installation
instructions](../benchmark/README.md#install-the-generators-once),
+then configure `conf/doris-cluster.conf` and run:
+
+```bash
+./bin/run-tpcds.sh -s 1 -d tpcds_sf1
+```
+
+This streams generated rows directly into Doris, creates the existing benchmark
+tables, collects statistics, and records three executions per query. It
supports
+SF1, SF100, SF1000, and SF10000. Preparation requires a new database.
+Use `--queries-only` to measure existing data again, with the same scale
factor.
+
+See [shared workflow options, result files, and timing
semantics](../benchmark/README.md).
+TPCDS uses the original scale-specific table DDL and multi-table queries.
+Install the plugin before running the command.
+
+# Original file-based workflow
+
## Usage
These scripts are used to make tpc-ds test.
diff --git a/tools/tpcds-tools/bin/run-tpcds.sh
b/tools/tpcds-tools/bin/run-tpcds.sh
new file mode 100755
index 00000000000..38e6ba69101
--- /dev/null
+++ b/tools/tpcds-tools/bin/run-tpcds.sh
@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+# 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.
+
+set -Eeuo pipefail
+SCRIPT_ROOT=$(cd "$(dirname "$0")" && pwd)
+exec bash "${SCRIPT_ROOT}/../../benchmark/run-benchmark.sh" tpcds "$@"
diff --git a/tools/tpch-tools/.gitignore b/tools/tpch-tools/.gitignore
new file mode 100644
index 00000000000..fbca2253799
--- /dev/null
+++ b/tools/tpch-tools/.gitignore
@@ -0,0 +1 @@
+results/
diff --git a/tools/tpch-tools/README.md b/tools/tpch-tools/README.md
index a7deac387ba..32a3ceb89f6 100644
--- a/tools/tpch-tools/README.md
+++ b/tools/tpch-tools/README.md
@@ -17,6 +17,27 @@ specific language governing permissions and limitations
under the License.
-->
+# Run TPCH with one command
+
+Install the Trino TPCH generator on every FE and BE using the
+[shared installation
instructions](../benchmark/README.md#install-the-generators-once),
+then configure `conf/doris-cluster.conf` and run:
+
+```bash
+./bin/run-tpch.sh -s 1 -d tpch_sf1
+```
+
+This streams generated rows directly into Doris, creates the existing benchmark
+tables, collects statistics, and records three executions per query. It
supports
+SF1, SF100, SF1000, and SF10000. Preparation requires a new database.
+Use `--queries-only` to measure existing data again, with the same scale
factor.
+
+See [shared workflow options, result files, and timing
semantics](../benchmark/README.md).
+TPCH uses the original scale-specific table DDL and multi-table queries.
+Install the plugin before running the command.
+
+# Original file-based workflow
+
## Usage
These scripts are used to make tpc-h test.
diff --git a/tools/tpch-tools/bin/run-tpch.sh b/tools/tpch-tools/bin/run-tpch.sh
new file mode 100755
index 00000000000..9f342c553cc
--- /dev/null
+++ b/tools/tpch-tools/bin/run-tpch.sh
@@ -0,0 +1,21 @@
+#!/usr/bin/env bash
+# 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.
+
+set -Eeuo pipefail
+SCRIPT_ROOT=$(cd "$(dirname "$0")" && pwd)
+exec bash "${SCRIPT_ROOT}/../../benchmark/run-benchmark.sh" tpch "$@"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
