bench: add spill_views SQL benchmarks for spilling StringView/BinaryView columns

benchmarks/bench.sh

@@ -112,6 +112,10 @@ parquet_row_filter_skip: Per-RG fully-matched RowFilter skip on Parquet (apache/
                           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)
+spill_views:            Sort and GROUP BY queries that spill StringView/BinaryView columns read from Parquet
+                          (https://github.com/apache/datafusion/issues/23564); each query sets its own memory limit
+                          (subgroups via BENCH_SUBGROUP: repeated = low-cardinality payload, distinct = all-distinct strings)
+                          (data generated inline by the suite's load SQL)
 null_aware_join:        Null-aware (NOT IN) hash join micro-benchmarks: uncorrelated, non-equality-correlated and equality-correlated
                           NOT IN across NULL fractions, to measure the per-pair join-filter work the correlated cases do
                           (data generated inline by the suite's load SQL from range(); knobs: NAJ_ROWS, NAJ_LARGE_ROWS)
@@ -279,6 +283,10 @@ main() {
                     # Data is generated inline by the suite's load SQL (COPY).
                     echo "parquet_row_filter_skip: no external data to generate"
                     ;;
+                spill_views)
+                    # Data is generated inline by the suite's load SQL (COPY).
+                    echo "spill_views: no external data to generate"
+                    ;;
                 null_aware_join)
                     # Data is generated inline by the suite's load SQL from range().
                     echo "null_aware_join: no external data to generate"
@@ -535,6 +543,9 @@ main() {
                 parquet_row_filter_skip)
                     run_parquet_row_filter_skip
                     ;;
+                spill_views)
+                    run_spill_views
+                    ;;
                 null_aware_join)
                     run_null_aware_join
                     ;;
@@ -998,6 +1009,20 @@ run_projection_subquery() {
       bash -c "$SQL_CARGO_COMMAND"
 }
 
+# Runs the spill_views suite: sort and GROUP BY queries that spill StringView
+# and BinaryView columns read from Parquet
+# (https://github.com/apache/datafusion/issues/23564). The load SQL COPYs 1M-row
+# Parquet files inline, so there is no data step. Each query sets its own memory
+# limit and target_partitions, so it spills regardless of the environment.
+#   BENCH_SUBGROUP  run one subgroup (repeated, distinct)
+run_spill_views() {
+    echo "Running spill_views benchmark (subgroup=${BENCH_SUBGROUP:-all})..."
+    debug_run env BENCH_NAME=spill_views \
+      ${BENCH_SUBGROUP:+BENCH_SUBGROUP="${BENCH_SUBGROUP}"} \
+      ${QUERY:+BENCH_QUERY="${QUERY}"}  \
+      bash -c "$SQL_CARGO_COMMAND"
+}
+
 # Runs the tpch in memory (needs tpch parquet data)
 run_tpch_mem() {
     SCALE_FACTOR=$1

benchmarks/sql_benchmarks/README.md

@@ -47,6 +47,7 @@ in the community:
 | `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`, `nulls`. The suite sets no engine config of its own, so by default it measures DataFusion's built-in left-deep `AND` short-circuit; point it at a system under test by exporting that system's own DataFusion setting (the harness builds its `SessionConfig` with `SessionConfig::from_env`). Every query is a `count(*)`, and the counts are checked in under `predicate_eval/results/`, so `--result-mode validate` also checks that a reordering under test still returns the same rows; the checked-in counts were persisted at the suite defaults (`PRED_ROWS=1000000`, `PRED_FILL=30`), so validation assumes those (the `scale` and `width` subgroups pin their own values per query and validate at any setting). |
 | `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`. |
+| `spill_views` | Sort and `GROUP BY` queries that spill `StringView` and `BinaryView` columns read from Parquet ([#23564](https://github.com/apache/datafusion/issues/23564)). Each query sets its own memory limit and `target_partitions = 4`, so it spills in any environment. Subgroups (`--subgroup`): `repeated` (sort with a low-cardinality payload that shares one dictionary buffer, where spill compaction can remove data), `distinct` (sort and `GROUP BY` on all-distinct strings, where compaction cannot remove data). |
 | `projection_subquery` | `IN`, `NOT IN` and `EXISTS` subqueries that sit in the `SELECT` list instead of a filter ([#25341](https://github.com/apache/datafusion/issues/25341)). Every query projects the boolean subquery result and aggregates it, so the measured cost is the decorrelation plan and not the size of the output. `q07` correlates on `<` instead of `=`, so it keeps the nested-loop plan and is the control that must not change. Size the two inline tables with `PSQ_ROWS` (default `30000`). The counts are checked in under `projection_subquery/results/`, so `--result-mode validate` also proves that a change to the decorrelation still returns the same rows; the checked-in counts were persisted at the default `PSQ_ROWS`, so validation assumes that value. |
 
 # Running Benchmarks

benchmarks/sql_benchmarks/spill_views/benchmarks/distinct/q04.benchmark

@@ -0,0 +1,9 @@
+subgroup distinct
+
+template sql_benchmarks/spill_views/spill_views.benchmark.template
+QPAD=04
+DATASET=distinct
+NAME=q04_sort_string_all_distinct
+SPILL_LIMIT=96M
+VIEW_TYPE=Utf8View
+DISTINCT_S=1000000

benchmarks/sql_benchmarks/spill_views/benchmarks/distinct/q05.benchmark

@@ -0,0 +1,9 @@
+subgroup distinct
+
+template sql_benchmarks/spill_views/spill_views.benchmark.template
+QPAD=05
+DATASET=distinct
+NAME=q05_group_by_string_all_distinct
+SPILL_LIMIT=96M
+VIEW_TYPE=Utf8View
+DISTINCT_S=1000000

benchmarks/sql_benchmarks/spill_views/benchmarks/repeated/q01.benchmark

@@ -0,0 +1,9 @@
+subgroup repeated
+
+template sql_benchmarks/spill_views/spill_views.benchmark.template
+QPAD=01
+DATASET=low_card_1
+NAME=q01_sort_string_1_distinct
+SPILL_LIMIT=40M
+VIEW_TYPE=Utf8View
+DISTINCT_S=1

benchmarks/sql_benchmarks/spill_views/benchmarks/repeated/q02.benchmark

@@ -0,0 +1,9 @@
+subgroup repeated
+
+template sql_benchmarks/spill_views/spill_views.benchmark.template
+QPAD=02
+DATASET=low_card_1000
+NAME=q02_sort_string_1000_distinct
+SPILL_LIMIT=40M
+VIEW_TYPE=Utf8View
+DISTINCT_S=1000

benchmarks/sql_benchmarks/spill_views/benchmarks/repeated/q03.benchmark

@@ -0,0 +1,9 @@
+subgroup repeated
+
+template sql_benchmarks/spill_views/spill_views.benchmark.template
+QPAD=03
+DATASET=low_card_1000_binary
+NAME=q03_sort_binary_1000_distinct
+SPILL_LIMIT=40M
+VIEW_TYPE=BinaryView
+DISTINCT_S=1000

benchmarks/sql_benchmarks/spill_views/init/cleanup.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS t;

benchmarks/sql_benchmarks/spill_views/init/settings.sql

@@ -0,0 +1,5 @@
+-- Session settings for the spill_views suite. `init` runs after `load`, so
+-- the COPY in the load script runs without the memory limit.
+set datafusion.catalog.information_schema = true;
+set datafusion.execution.target_partitions = 4;
+set datafusion.runtime.memory_limit = '${SPILL_LIMIT}';

benchmarks/sql_benchmarks/spill_views/load/distinct.sql

@@ -0,0 +1,18 @@
+-- All-distinct string values. The Parquet dictionary falls back to plain
+-- encoding, and there is no repeated data for spill compaction to remove.
+--
+-- 1M rows. `id` holds distinct values in shuffled order (7919 is
+-- invertible modulo the prime 1000003), so ORDER BY id must really sort.
+-- `s` is a 64-byte payload.
+COPY (
+  SELECT
+    (value * 7919) % 1000003 AS id,
+    'payload-' || lpad(CAST(value AS VARCHAR), 56, '0') AS s
+  FROM generate_series(1, 1000000)
+)
+TO 'sql_benchmarks/spill_views/scratch/distinct.parquet'
+STORED AS PARQUET;
+
+CREATE EXTERNAL TABLE t
+STORED AS PARQUET
+LOCATION 'sql_benchmarks/spill_views/scratch/distinct.parquet';

benchmarks/sql_benchmarks/spill_views/load/low_card_1.sql

@@ -0,0 +1,18 @@
+-- One distinct string value. Parquet dictionary-encodes it, so all views
+-- point into one small shared buffer.
+--
+-- 1M rows. `id` holds distinct values in shuffled order (7919 is
+-- invertible modulo the prime 1000003), so ORDER BY id must really sort.
+-- `s` is a 64-byte payload.
+COPY (
+  SELECT
+    (value * 7919) % 1000003 AS id,
+    'payload-' || lpad('x', 56, '0') AS s
+  FROM generate_series(1, 1000000)
+)
+TO 'sql_benchmarks/spill_views/scratch/low_card_1.parquet'
+STORED AS PARQUET;
+
+CREATE EXTERNAL TABLE t
+STORED AS PARQUET
+LOCATION 'sql_benchmarks/spill_views/scratch/low_card_1.parquet';

benchmarks/sql_benchmarks/spill_views/load/low_card_1000.sql

@@ -0,0 +1,18 @@
+-- 1000 distinct string values (64 KB dictionary). Parquet dictionary-encodes
+-- them, so all views point into one shared buffer.
+--
+-- 1M rows. `id` holds distinct values in shuffled order (7919 is
+-- invertible modulo the prime 1000003), so ORDER BY id must really sort.
+-- `s` is a 64-byte payload.
+COPY (
+  SELECT
+    (value * 7919) % 1000003 AS id,
+    'payload-' || lpad(CAST(value % 1000 AS VARCHAR), 56, '0') AS s
+  FROM generate_series(1, 1000000)
+)
+TO 'sql_benchmarks/spill_views/scratch/low_card_1000.parquet'
+STORED AS PARQUET;
+
+CREATE EXTERNAL TABLE t
+STORED AS PARQUET
+LOCATION 'sql_benchmarks/spill_views/scratch/low_card_1000.parquet';

benchmarks/sql_benchmarks/spill_views/load/low_card_1000_binary.sql

@@ -0,0 +1,18 @@
+-- Same as low_card_1000, but stored as binary, so it is read back as
+-- BinaryView.
+--
+-- 1M rows. `id` holds distinct values in shuffled order (7919 is
+-- invertible modulo the prime 1000003), so ORDER BY id must really sort.
+-- `s` is a 64-byte payload.
+COPY (
+  SELECT
+    (value * 7919) % 1000003 AS id,
+    CAST('payload-' || lpad(CAST(value % 1000 AS VARCHAR), 56, '0') AS BYTEA) AS s
+  FROM generate_series(1, 1000000)
+)
+TO 'sql_benchmarks/spill_views/scratch/low_card_1000_binary.parquet'
+STORED AS PARQUET;
+
+CREATE EXTERNAL TABLE t
+STORED AS PARQUET
+LOCATION 'sql_benchmarks/spill_views/scratch/low_card_1000_binary.parquet';

benchmarks/sql_benchmarks/spill_views/queries/q01.sql

@@ -0,0 +1,4 @@
+-- Sort 1M rows by a shuffled key with a 64-byte string payload that has one
+-- distinct value. The sort spills, and each spilled batch holds views into
+-- the shared dictionary buffer.
+SELECT id, s FROM t ORDER BY id;

benchmarks/sql_benchmarks/spill_views/queries/q02.sql

@@ -0,0 +1,3 @@
+-- Sort 1M rows by a shuffled key with a 64-byte string payload that has 1000
+-- distinct values (a 64 KB dictionary buffer shared by all views).
+SELECT id, s FROM t ORDER BY id;

benchmarks/sql_benchmarks/spill_views/queries/q03.sql

@@ -0,0 +1,2 @@
+-- Same as q02, with a BinaryView payload instead of a StringView payload.
+SELECT id, s FROM t ORDER BY id;

benchmarks/sql_benchmarks/spill_views/queries/q04.sql

@@ -0,0 +1,3 @@
+-- Sort 1M rows by a shuffled key with an all-distinct 64-byte string payload.
+-- Spill compaction cannot remove data here, so this measures its cost.
+SELECT id, s FROM t ORDER BY id;

benchmarks/sql_benchmarks/spill_views/queries/q05.sql

@@ -0,0 +1,4 @@
+-- GROUP BY an all-distinct 64-byte string key. The final aggregation spills
+-- its group keys, and the keys in each spilled batch are distinct, so spill
+-- compaction cannot remove data here: this measures its cost.
+SELECT s, count(*) AS c FROM t GROUP BY s;

benchmarks/sql_benchmarks/spill_views/scratch/.gitignore

@@ -0,0 +1 @@
+*.parquet

benchmarks/sql_benchmarks/spill_views/spill_views.benchmark.template

@@ -0,0 +1,53 @@
+# Shared template for the spill_views suite. Each qNN.benchmark sets its
+# `subgroup` directive and includes this template with parameters:
+#   QPAD          zero-padded query id / query file stem   (e.g. 01)
+#   DATASET       load script stem under load/             (e.g. low_card_1)
+#   NAME          criterion display name
+#   SPILL_LIMIT   memory limit for the measured query      (e.g. 40M)
+#   VIEW_TYPE     expected Arrow type of column `s`        (Utf8View, BinaryView)
+#   DISTINCT_S    expected number of distinct values of `s`
+#
+# The load script COPYs a 1M-row Parquet file into scratch/ and registers it
+# as table `t` (id BIGINT in shuffled order, s = a 64-byte payload). Reading
+# the data back from Parquet is load-bearing: the Parquet reader makes the
+# views of a dictionary-encoded column point into one shared buffer, which is
+# the case that spill compaction targets.
+#
+# init/settings.sql sets the memory limit and pins target_partitions = 4. The
+# limits were sized for 4 partitions so that each query spills on main
+# without running out of memory. Do not remove the pin: with more partitions
+# the per-partition sort reservations can exceed the limit.
+
+load sql_benchmarks/spill_views/load/${DATASET}.sql
+
+init sql_benchmarks/spill_views/init/settings.sql
+
+name ${NAME}
+group spill_views
+
+assert I
+SELECT count(*) = 1000000 FROM t;
+----
+true
+
+assert I
+SELECT count(DISTINCT s) = ${DISTINCT_S} FROM t;
+----
+true
+
+# The column must be read as a view type, else the suite measures nothing.
+assert I
+SELECT arrow_typeof(s) = '${VIEW_TYPE}' FROM t LIMIT 1;
+----
+true
+
+# Guard against silent config no-ops: the query must run under the limit.
+assert II
+SELECT name, value FROM information_schema.df_settings WHERE name IN ('datafusion.runtime.memory_limit', 'datafusion.execution.target_partitions') ORDER BY name;
+----
+datafusion.execution.target_partitions|4
+datafusion.runtime.memory_limit|${SPILL_LIMIT}
+
+run sql_benchmarks/spill_views/queries/q${QPAD}.sql
+
+cleanup sql_benchmarks/spill_views/init/cleanup.sql

benchmarks/sql_benchmarks/spill_views/spill_views.suite

@@ -0,0 +1,11 @@
+description = "Sort and aggregation queries that spill StringView and BinaryView columns read from Parquet (https://github.com/apache/datafusion/issues/23564). Each query sets its own memory limit and target_partitions = 4, so it spills with any environment. The repeated subgroup sorts rows whose string payload has few distinct values, so the Parquet reader makes views into one shared dictionary buffer and the spill writer can compact that buffer. The distinct subgroup sorts and groups all-distinct strings, where compaction cannot remove data and only adds work."
+
+query_pattern = "q{QUERY_ID_PADDED}.benchmark"
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- spill_views"
+description = "Run all spill_views queries."
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- spill_views --subgroup distinct"
+description = "Run only the all-distinct (cost case) queries."