diff --git a/benchmarks/bench.sh b/benchmarks/bench.sh index 52b78c844a73a..df9b7f6c94f16 100755 --- a/benchmarks/bench.sh +++ b/benchmarks/bench.sh @@ -108,6 +108,10 @@ predicate_eval: Conjunctive (AND) filter-evaluation micro-benchmarks; ea adaptive predicate-ordering system behaves across them (see https://github.com/apache/datafusion/issues/11262) (subgroups via BENCH_SUBGROUP: costsel, cost, selectivity, cardinality, width, scale, neutral, correlation, drift) (toggle a system under test with its native DATAFUSION_* env var; size data with PRED_ROWS, string width with PRED_FILL) +parquet_row_filter_skip: Per-RG fully-matched RowFilter skip on Parquet (apache/datafusion#23696); clustered string key + low-selectivity + range filter + pushdown, so most row groups are fully matched and the per-row RowFilter is skipped on them + (subgroups via BENCH_SUBGROUP: skip = clustered key so the skip fires, control = scrambled key so it never fires) + (data generated inline by the suite's load SQL; knobs: PRED_ROWS, RG_SIZE) # ClickBench Benchmarks clickbench_1: ClickBench queries against a single parquet file @@ -255,6 +259,10 @@ main() { # Data is generated inline by the suite's load SQL. echo "predicate_eval: no external data to generate" ;; + parquet_row_filter_skip) + # Data is generated inline by the suite's load SQL (COPY). + echo "parquet_row_filter_skip: no external data to generate" + ;; tpcds) data_tpcds ;; @@ -475,6 +483,9 @@ main() { predicate_eval) run_predicate_eval ;; + parquet_row_filter_skip) + run_parquet_row_filter_skip + ;; tpcds) run_tpcds ;; @@ -844,6 +855,27 @@ run_predicate_eval() { bash -c "$SQL_CARGO_COMMAND" } +# Runs the parquet_row_filter_skip suite: the load SQL COPYs a Parquet file +# inline (fixed-width string key, ordered so each row group holds a disjoint +# sorted range) and the query applies a low-selectivity range filter, so all +# but the first row group is fully matched by statistics and the per-row +# RowFilter is skipped on them (apache/datafusion#23696). The control subgroup +# scrambles the key so no row group is ever fully matched, measuring the +# overhead of the check when it cannot fire. Data is inline, so no data step. +# Knobs (string-substituted into the load SQL, not engine config): +# BENCH_SUBGROUP run one subgroup (skip, control) +# PRED_ROWS synthetic row count (default 10_000_000) +# RG_SIZE parquet row-group size (default 1_000_000) +run_parquet_row_filter_skip() { + echo "Running parquet_row_filter_skip benchmark (subgroup=${BENCH_SUBGROUP:-all}, rows=${PRED_ROWS:-10000000}, rg_size=${RG_SIZE:-1000000})..." + debug_run env BENCH_NAME=parquet_row_filter_skip \ + ${BENCH_SUBGROUP:+BENCH_SUBGROUP="${BENCH_SUBGROUP}"} \ + PRED_ROWS="${PRED_ROWS:-10000000}" \ + RG_SIZE="${RG_SIZE:-1000000}" \ + ${QUERY:+BENCH_QUERY="${QUERY}"} \ + bash -c "$SQL_CARGO_COMMAND" +} + # Runs the tpch in memory (needs tpch parquet data) run_tpch_mem() { SCALE_FACTOR=$1 diff --git a/benchmarks/sql_benchmarks/README.md b/benchmarks/sql_benchmarks/README.md index dfb09e0a3a4a2..1ce1fa488c6b8 100644 --- a/benchmarks/sql_benchmarks/README.md +++ b/benchmarks/sql_benchmarks/README.md @@ -44,6 +44,7 @@ in the community: | `tpch` | TPC‑H queries | | `wide_schema` | Small-projection queries on a wide (1024-col, 256-file) synthetic dataset; runs `wide` + `narrow` subgroups for comparison | | `predicate_eval` | Conjunctive (AND) filter-evaluation micro-benchmarks; each subgroup is a different predicate pattern, to test how an adaptive predicate-ordering system behaves across them ([#11262](https://github.com/apache/datafusion/issues/11262)). Subgroups (`--subgroup`): `costsel`, `cost`, `selectivity`, `cardinality`, `width`, `scale`, `neutral`, `correlation`, `drift`. Configure the system under test through its DataFusion settings. | +| `parquet_row_filter_skip` | Micro-benchmark for the per-row-group fully-matched RowFilter skip on Parquet scans ([#23696](https://github.com/apache/datafusion/issues/23696)). Subgroups (`--subgroup`): `skip` (clustered key, most row groups fully matched by statistics so the per-row filter is skipped), `control` (scrambled key, no row group is ever fully matched). Size the data with `PRED_ROWS` and the row-group size with `RG_SIZE`. | # Running Benchmarks @@ -168,8 +169,9 @@ Some benchmarks use custom environment variables as outlined below: | BENCH_SORTED | Used in the sort_tpch benchmark to indicate whether the lineitem table should be sorted. | false | | SORTED_BY | Used in the clickbench_sorted benchmark to indicate the column to sort by. | `EventTime` | | SORTED_ORDER | Used in the clickbench_sorted benchmark to indicate the sort order of the column. | `ASC` | -| PRED_ROWS | Used in the predicate_eval benchmark to size the synthetic table (the `scale` subgroup overrides this per query). | `1000000` | +| PRED_ROWS | Used in the predicate_eval benchmark to size the synthetic table (the `scale` subgroup overrides this per query), and in the parquet_row_filter_skip benchmark to size the generated Parquet datasets (default `10000000` there). | `1000000` | | PRED_FILL | Used in the predicate_eval benchmark as the string-column width knob (filler chars per marker). | `30` | +| RG_SIZE | Used in the parquet_row_filter_skip benchmark as the Parquet `max_row_group_size` for the generated datasets. | `1000000` | ## How it works diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/control/q10.benchmark b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/control/q10.benchmark new file mode 100644 index 0000000000000..4c64baa83300e --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/control/q10.benchmark @@ -0,0 +1,7 @@ +subgroup control + +template sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template +SUBGROUP=control +QPAD=10 +DATASET=scrambled +NAME=control_q10_scrambled_never_matched diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q01.benchmark b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q01.benchmark new file mode 100644 index 0000000000000..825605c6bb65e --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q01.benchmark @@ -0,0 +1,7 @@ +subgroup skip + +template sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template +SUBGROUP=skip +QPAD=01 +DATASET=clustered +NAME=skip_q01_string_range_low_selectivity diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q02.benchmark b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q02.benchmark new file mode 100644 index 0000000000000..30f1ff8b48181 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q02.benchmark @@ -0,0 +1,7 @@ +subgroup skip + +template sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template +SUBGROUP=skip +QPAD=02 +DATASET=clustered +NAME=skip_q02_string_range_all_matched diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q03.benchmark b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q03.benchmark new file mode 100644 index 0000000000000..12015067f8b32 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/benchmarks/skip/q03.benchmark @@ -0,0 +1,7 @@ +subgroup skip + +template sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template +SUBGROUP=skip +QPAD=03 +DATASET=clustered +NAME=skip_q03_filter_column_projected diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/init/cleanup.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/init/cleanup.sql new file mode 100644 index 0000000000000..48f076a9fa652 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/init/cleanup.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS t; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/init/settings.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/init/settings.sql new file mode 100644 index 0000000000000..0f084fae938df --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/init/settings.sql @@ -0,0 +1,9 @@ +-- Session settings for the parquet_row_filter_skip suite. `init` runs after +-- `load` and before the asserts and the benchmarked query, so these apply to +-- the measured scan (the COPY in the load script does not need them). +-- +-- information_schema is enabled so the template can assert that +-- pushdown_filters actually took effect: without pushdown there is no +-- RowFilter to skip and the suite would silently measure nothing. +set datafusion.catalog.information_schema = true; +set datafusion.execution.parquet.pushdown_filters = true; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/load/clustered.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/load/clustered.sql new file mode 100644 index 0000000000000..dd6bdfc42eb20 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/load/clustered.sql @@ -0,0 +1,44 @@ +-- Clustered Parquet dataset for the fully-matched RowFilter-skip benchmark. +-- +-- `skey` is a fixed-width, zero-padded, monotonically increasing string, so +-- each row group holds a disjoint, sorted range of keys. With +-- `pushdown_filters=true` (set in init/settings.sql), a low-selectivity range +-- predicate (see the queries) leaves the first row group straddling and every +-- later row group fully matched by min/max statistics, which is exactly what +-- the per-RG RowFilter skip targets. +-- +-- The ORDER BY is load-bearing: the whole benchmark rests on the file being +-- written in key order so that row-group min/max ranges are disjoint. Without +-- it the write order is at the mercy of the physical planner (a round-robin +-- repartition + coalesce would silently interleave batches and destroy the +-- clustering, leaving nothing to measure). +-- +-- Knobs: PRED_ROWS (row count, must exceed the 100_000 predicate cutoff), +-- RG_SIZE (parquet row-group size). +COPY ( + SELECT + lpad(CAST(value AS VARCHAR), 10, '0') AS skey, + (value * 7) % 1000000 AS p0, + (value * 13) % 1000000 AS p1, + (value * 17) % 1000000 AS p2, + (value * 19) % 1000000 AS p3, + (value * 23) % 1000000 AS p4, + (value * 29) % 1000000 AS p5, + (value * 31) % 1000000 AS p6, + (value * 37) % 1000000 AS p7, + (value * 41) % 1000000 AS p8, + (value * 43) % 1000000 AS p9, + (value * 47) % 1000000 AS p10, + (value * 53) % 1000000 AS p11, + (value * 59) % 1000000 AS p12, + (value * 61) % 1000000 AS p13 + FROM generate_series(1, ${PRED_ROWS:-10000000}) + ORDER BY value +) +TO 'sql_benchmarks/parquet_row_filter_skip/scratch/clustered.parquet' +STORED AS PARQUET +OPTIONS ('format.max_row_group_size' '${RG_SIZE:-1000000}'); + +CREATE EXTERNAL TABLE t +STORED AS PARQUET +LOCATION 'sql_benchmarks/parquet_row_filter_skip/scratch/clustered.parquet'; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/load/scrambled.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/load/scrambled.sql new file mode 100644 index 0000000000000..c8c0e0bf97822 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/load/scrambled.sql @@ -0,0 +1,43 @@ +-- Scrambled control dataset for the fully-matched RowFilter-skip benchmark. +-- +-- Same schema, key domain, and payload as clustered.sql, but `skey` is a +-- pseudo-random permutation of 1..PRED_ROWS (999983 is prime, so the map +-- `v -> (v * 999983) % PRED_ROWS + 1` is a bijection whenever PRED_ROWS is +-- not a multiple of 999983 -- true for the default and any power-of-ten +-- size). Every row group therefore spans nearly the whole key range and no +-- row group is ever fully matched by min/max statistics: the per-row +-- RowFilter must run everywhere. Queries over this dataset measure the +-- overhead of the fully-matched check when it can never fire. +-- +-- Keeping the domain exactly 1..PRED_ROWS lets the template's row-count +-- asserts hold for both datasets. +-- +-- Knobs: PRED_ROWS (row count, must exceed the 100_000 predicate cutoff), +-- RG_SIZE (parquet row-group size). +COPY ( + SELECT + lpad(CAST((value * 999983) % ${PRED_ROWS:-10000000} + 1 AS VARCHAR), 10, '0') AS skey, + (value * 7) % 1000000 AS p0, + (value * 13) % 1000000 AS p1, + (value * 17) % 1000000 AS p2, + (value * 19) % 1000000 AS p3, + (value * 23) % 1000000 AS p4, + (value * 29) % 1000000 AS p5, + (value * 31) % 1000000 AS p6, + (value * 37) % 1000000 AS p7, + (value * 41) % 1000000 AS p8, + (value * 43) % 1000000 AS p9, + (value * 47) % 1000000 AS p10, + (value * 53) % 1000000 AS p11, + (value * 59) % 1000000 AS p12, + (value * 61) % 1000000 AS p13 + FROM generate_series(1, ${PRED_ROWS:-10000000}) + ORDER BY value +) +TO 'sql_benchmarks/parquet_row_filter_skip/scratch/scrambled.parquet' +STORED AS PARQUET +OPTIONS ('format.max_row_group_size' '${RG_SIZE:-1000000}'); + +CREATE EXTERNAL TABLE t +STORED AS PARQUET +LOCATION 'sql_benchmarks/parquet_row_filter_skip/scratch/scrambled.parquet'; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template b/benchmarks/sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template new file mode 100644 index 0000000000000..44c776f5f4ac2 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.benchmark.template @@ -0,0 +1,50 @@ +# Shared template for the parquet_row_filter_skip suite. Each qNN.benchmark sets +# its `subgroup` directive and includes this template with parameters: +# SUBGROUP subgroup name, also the query sub-directory (skip, control) +# QPAD zero-padded query id / query file stem (e.g. 01) +# DATASET load script stem under load/ (clustered, scrambled) +# NAME criterion display name +# Optional (consumed by the load script via ${...:-default}): +# PRED_ROWS synthetic row count (default 10_000_000) +# RG_SIZE parquet max row-group size (default 1_000_000) +# +# The load script COPYs the Parquet dataset and init/settings.sql enables +# `pushdown_filters` so the fully-matched RowFilter-skip path is exercised. +# The table is always named `t`, so the asserts and cleanup are uniform. +# +# The queries share a fixed cutoff key of '0000100000' (row 100_000), so +# PRED_ROWS must stay above 100_000; the second assert fails loudly otherwise. +# Both datasets use the key domain 1..PRED_ROWS, so the asserts hold for both. + +load sql_benchmarks/parquet_row_filter_skip/load/${DATASET}.sql + +init sql_benchmarks/parquet_row_filter_skip/init/settings.sql + +name ${NAME} +group parquet_row_filter_skip + +# The dataset has exactly PRED_ROWS rows. +assert I +SELECT count(*) = ${PRED_ROWS:-10000000} FROM t; +---- +true + +# The cutoff predicate matches exactly PRED_ROWS - 99_999 rows. This doubles +# as a correctness canary for the optimization under test: a RowFilter skip +# that fires on a not-fully-matched row group returns extra rows and trips +# this assert. +assert I +SELECT count(*) = ${PRED_ROWS:-10000000} - 99999 FROM t WHERE skey >= '0000100000'; +---- +true + +# Guard against silent config no-ops: without pushdown_filters there is no +# RowFilter at all and the suite measures nothing. +assert I +SELECT value = 'true' FROM information_schema.df_settings WHERE name = 'datafusion.execution.parquet.pushdown_filters'; +---- +true + +run sql_benchmarks/parquet_row_filter_skip/queries/${SUBGROUP}/q${QPAD}.sql + +cleanup sql_benchmarks/parquet_row_filter_skip/init/cleanup.sql diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.suite b/benchmarks/sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.suite new file mode 100644 index 0000000000000..1cf46dd2ef188 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/parquet_row_filter_skip.suite @@ -0,0 +1,26 @@ +description = "Micro-benchmark for the per-row-group fully-matched RowFilter skip on Parquet scans (apache/datafusion#23696). Builds a clustered Parquet file whose fixed-width string key is monotonically increasing, so each row group holds a disjoint sorted range. With filter pushdown enabled, a low-selectivity range predicate leaves the first row group straddling and the rest fully matched by statistics, so the decoder can skip the per-row RowFilter (and decoding the filter column) on the fully-matched run. The control subgroup runs the same predicate over a scrambled key, where no row group is ever fully matched, to measure the overhead of the check when it cannot fire. Size the data with PRED_ROWS and the row-group size with RG_SIZE." + +query_pattern = "q{QUERY_ID_PADDED}.benchmark" + +[[options]] +name = "rows" +short = "r" +env = "PRED_ROWS" +default = "10000000" +values = ["10000000", "..."] +help = "Sets the number of rows in the generated Parquet datasets (must be greater than 100000, the fixed predicate cutoff)." + +[[options]] +name = "rg-size" +env = "RG_SIZE" +default = "1000000" +values = ["1000000", "..."] +help = "Sets the Parquet max_row_group_size used when writing the generated datasets." + +[[examples]] +command = "cargo run --release --bin benchmark_runner -- parquet_row_filter_skip" +description = "Run every subgroup with the default 10M rows and 1M-row row groups." + +[[examples]] +command = "cargo run --release --bin benchmark_runner -- parquet_row_filter_skip --subgroup skip --rg-size 100000" +description = "Run only the skip subgroup with 100 row groups instead of 10." diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/control/q10.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/control/q10.sql new file mode 100644 index 0000000000000..a5950e925b0ce --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/control/q10.sql @@ -0,0 +1,8 @@ +-- Control: the same cutoff predicate as skip/q01 but over the scrambled +-- dataset, where every row group spans nearly the whole key range, so no row +-- group is ever fully matched and the per-row RowFilter runs everywhere. +-- Measures the overhead of the fully-matched check when it cannot fire; the +-- skip optimization should be performance-neutral here. +SELECT sum(p0)+sum(p1)+sum(p2)+sum(p3)+sum(p4)+sum(p5)+sum(p6)+sum(p7)+sum(p8)+sum(p9)+sum(p10)+sum(p11)+sum(p12)+sum(p13) AS s +FROM t +WHERE skey >= '0000100000'; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q01.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q01.sql new file mode 100644 index 0000000000000..33468c19226fa --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q01.sql @@ -0,0 +1,7 @@ +-- Low-selectivity range filter on the clustered string key: every row group +-- except the first is fully matched by statistics, so the per-row RowFilter is +-- skipped on those RGs. `skey` is not projected, so the skip also avoids +-- decoding the filter column on the fully-matched run. +SELECT sum(p0)+sum(p1)+sum(p2)+sum(p3)+sum(p4)+sum(p5)+sum(p6)+sum(p7)+sum(p8)+sum(p9)+sum(p10)+sum(p11)+sum(p12)+sum(p13) AS s +FROM t +WHERE skey >= '0000100000'; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q02.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q02.sql new file mode 100644 index 0000000000000..a8eb2d6ca08f1 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q02.sql @@ -0,0 +1,6 @@ +-- Matches every row: the minimum key is '0000000001', so every row group -- +-- including the first -- is fully matched by statistics and the per-row +-- RowFilter is skipped everywhere. The upper bound for the optimization. +SELECT sum(p0)+sum(p1)+sum(p2)+sum(p3)+sum(p4)+sum(p5)+sum(p6)+sum(p7)+sum(p8)+sum(p9)+sum(p10)+sum(p11)+sum(p12)+sum(p13) AS s +FROM t +WHERE skey >= '0000000000'; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q03.sql b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q03.sql new file mode 100644 index 0000000000000..2879d222b240e --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/queries/skip/q03.sql @@ -0,0 +1,8 @@ +-- Same cutoff predicate as q01, but the filter column IS projected: the +-- fully-matched skip still avoids evaluating the per-row filter, but can no +-- longer avoid decoding `skey`. Measures the common case where the win is +-- smaller than q01's best case. +SELECT min(skey) AS first_key, + sum(p0)+sum(p1)+sum(p2)+sum(p3)+sum(p4)+sum(p5)+sum(p6)+sum(p7)+sum(p8)+sum(p9)+sum(p10)+sum(p11)+sum(p12)+sum(p13) AS s +FROM t +WHERE skey >= '0000100000'; diff --git a/benchmarks/sql_benchmarks/parquet_row_filter_skip/scratch/.gitignore b/benchmarks/sql_benchmarks/parquet_row_filter_skip/scratch/.gitignore new file mode 100644 index 0000000000000..4bed5da93fb28 --- /dev/null +++ b/benchmarks/sql_benchmarks/parquet_row_filter_skip/scratch/.gitignore @@ -0,0 +1 @@ +*.parquet