bench: add ASOF join benchmarks

benchmarks/README.md

@@ -923,6 +923,21 @@ cargo run --release --bin dfbench -- h2o --join-paths ./benchmarks/data/h2o/J1_1
 
 # Micro-Benchmarks
 
+## ASOF Join
+
+This benchmark exercises ASOF joins across relative input sizes, available
+input ordering, equality-group cardinality and skew, match direction, and
+payload width. One keyed workload reads pre-sorted Parquet inputs from
+`benchmarks/data/asof_join` with declared ordering so it measures the join
+without including an input sort.
+
+### Example Run
+
+```bash
+./bench.sh data asof_join
+./bench.sh run asof_join
+```
+
 ## Nested Loop Join
 
 This benchmark focuses on the performance of queries with nested loop joins, minimizing other overheads such as scanning data sources or evaluating predicates.

benchmarks/bench.sh

@@ -160,6 +160,7 @@ imdb:                   Join Order Benchmark (JOB) using the IMDB dataset conver
 
 # Micro-Benchmarks (specific operators and features)
 cancellation:           How long cancelling a query takes
+asof_join:              ASOF join workloads varying size, ordering, grouping, match direction, and payload width
 nlj:                    Benchmark for simple nested loop joins, testing various join scenarios
 hj:                     Benchmark for simple hash joins, testing various join scenarios
 smj:                    Benchmark for simple sort merge joins, testing various join scenarios
@@ -238,6 +239,7 @@ main() {
                     data_clickbench_1
                     data_clickbench_partitioned
                     data_imdb
+                    data_asof_join
                     # nlj uses range() function, no data generation needed
                     ;;
                 tpch)
@@ -269,6 +271,9 @@ main() {
                     # Data is generated inline by the suite's load SQL (COPY).
                     echo "parquet_row_filter_skip: no external data to generate"
                     ;;
+                asof_join)
+                    data_asof_join
+                    ;;
                 tpcds)
                     data_tpcds
                     ;;
@@ -479,6 +484,7 @@ main() {
                     run_h2o_join "BIG" "PARQUET" "join"
                     run_imdb
                     run_external_aggr
+                    run_asof_join
                     run_nlj
                     run_hj
                     run_tpcds
@@ -512,6 +518,9 @@ main() {
                 parquet_row_filter_skip)
                     run_parquet_row_filter_skip
                     ;;
+                asof_join)
+                    run_asof_join
+                    ;;
                 tpcds)
                     run_tpcds
                     ;;
@@ -1632,6 +1641,51 @@ run_topk_sorted_tpch() {
     $CARGO_COMMAND --bin dfbench -- sort-tpch --iterations 5 --path "${TPCH_DIR}" -o "${RESULTS_FILE}" --sorted --limit 100 ${QUERY_ARG} ${LATENCY_ARG}
 }
 
+# Generates the pre-sorted Parquet inputs for the ASOF join benchmark.
+data_asof_join() {
+    ASOF_DIR="${DATA_DIR}/asof_join"
+    LEFT_FILE="${ASOF_DIR}/q07_left.parquet"
+    RIGHT_FILE="${ASOF_DIR}/q07_right.parquet"
+
+    if [ -f "${LEFT_FILE}" ] && [ -f "${RIGHT_FILE}" ]; then
+        echo "ASOF join benchmark data already exists at ${ASOF_DIR}"
+        return
+    fi
+
+    mkdir -p "${ASOF_DIR}"
+    echo "Generating ASOF join benchmark data at ${ASOF_DIR}..."
+    (
+        cd "${DATAFUSION_DIR}"
+        debug_run $CARGO_COMMAND -p datafusion-cli -- -c "
+            COPY (
+                SELECT value / 10 AS group_key,
+                       value % 10 + 1 AS ts,
+                       value AS payload
+                FROM range(100000)
+                ORDER BY group_key, ts
+            )
+            TO '${LEFT_FILE}' STORED AS PARQUET;
+
+            COPY (
+                SELECT value / 10 AS group_key,
+                       value % 10 AS ts,
+                       value AS payload
+                FROM range(100000)
+                ORDER BY group_key, ts
+            )
+            TO '${RIGHT_FILE}' STORED AS PARQUET;
+        "
+    )
+}
+
+# Runs the ASOF join benchmark
+run_asof_join() {
+    RESULTS_FILE="${RESULTS_DIR}/asof_join.json"
+    echo "RESULTS_FILE: ${RESULTS_FILE}"
+    echo "Running ASOF join benchmark..."
+    debug_run $CARGO_COMMAND --bin benchmark_runner -- asof_join --iterations 5 --path "${DATA_DIR}" -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
+}
+
 # Runs the nlj benchmark
 run_nlj() {
     RESULTS_FILE="${RESULTS_DIR}/nlj.json"

benchmarks/sql_benchmarks/README.md

@@ -29,6 +29,7 @@ in the community:
 
 | Benchmark Suite       | Description                                                        |
 |-----------------------|--------------------------------------------------------------------|
+| `asof_join`           | ASOF join benchmarks across size, ordering, grouping, match direction, and payload width |
 | `clickbench`          | ClickBench benchmark                                               |
 | `clickbench extended` | 12 additional, more complex queries against the Clickbench dataset |
 | `clickbench_sorted`   | ClickBench benchmark using a pre-sorted hits file.                 |

benchmarks/sql_benchmarks/asof_join/asof_join.suite

@@ -0,0 +1,21 @@
+# Workloads vary these axes:
+# - relative left and right input sizes
+# - required ordering already available or included in the measured query
+# - no equality key, many balanced groups, or heavily skewed groups
+# - latest right row at or before, or earliest right row at or after, each left row
+# - narrow integer or wide UTF-8 payloads
+
+description = "ASOF join SQL benchmarks"
+
+query_pattern = "q{QUERY_ID_PADDED}.benchmark"
+
+[path_replacements]
+DATA_DIR = "../../data"
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- asof_join"
+description = "Run all ASOF join queries."
+
+[[examples]]
+command = "cargo run --release --bin benchmark_runner -- asof_join --query 7"
+description = "Run the pre-sorted keyed ASOF join query."

benchmarks/sql_benchmarks/asof_join/benchmarks/q01.benchmark

@@ -0,0 +1,17 @@
+name Q01
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left: 1M rows; right: 10K rows; no equality key; ascending timestamps.
+-- Matches the latest right row at or before each left timestamp (l.ts >= r.ts).
+WITH left_input AS (
+    SELECT value AS ts, value AS payload FROM range(1000000)
+),
+right_input AS (
+    SELECT value AS ts, value AS payload FROM range(10000)
+)
+SELECT l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts);

benchmarks/sql_benchmarks/asof_join/benchmarks/q02.benchmark

@@ -0,0 +1,25 @@
+name Q02
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left and right: 1M rows each; 10K balanced equality groups; integer payloads.
+-- Inputs are not ordered by (key, ts), so required ordering is included.
+-- Matches the latest right row at or before each left timestamp (l.ts >= r.ts).
+WITH left_input AS (
+    SELECT value % 10000 AS key,
+           value / 10000 + 1 AS ts,
+           value AS payload
+    FROM range(1000000)
+),
+right_input AS (
+    SELECT value % 10000 AS key,
+           value / 10000 AS ts,
+           value AS payload
+    FROM range(1000000)
+)
+SELECT l.key, l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts)
+ON l.key = r.key;

benchmarks/sql_benchmarks/asof_join/benchmarks/q03.benchmark

@@ -0,0 +1,26 @@
+name Q03
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left and right: 250K rows each; 10K balanced equality groups.
+-- Inputs are not ordered by (key, ts), so required ordering is included.
+-- Matches the latest right row at or before each left timestamp and materializes
+-- 256-byte UTF-8 payloads from both inputs.
+WITH left_input AS (
+    SELECT value % 10000 AS key,
+           value / 10000 + 1 AS ts,
+           repeat('x', 256) AS payload
+    FROM range(250000)
+),
+right_input AS (
+    SELECT value % 10000 AS key,
+           value / 10000 AS ts,
+           repeat('y', 256) AS payload
+    FROM range(250000)
+)
+SELECT l.key, l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts)
+ON l.key = r.key;

benchmarks/sql_benchmarks/asof_join/benchmarks/q04.benchmark

@@ -0,0 +1,18 @@
+name Q04
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left and right: 500K rows each; no equality key; ascending source timestamps.
+-- Matches the earliest right row at or after each left timestamp (l.ts <= r.ts),
+-- so the measured query includes the required descending ordering.
+WITH left_input AS (
+    SELECT value AS ts, value AS payload FROM range(500000)
+),
+right_input AS (
+    SELECT value AS ts, value AS payload FROM range(500000)
+)
+SELECT l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts <= r.ts);

benchmarks/sql_benchmarks/asof_join/benchmarks/q05.benchmark

@@ -0,0 +1,18 @@
+name Q05
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left: 100K rows; right: 1M rows; no equality key; ascending timestamps.
+-- Matches the latest right row at or before each left timestamp (l.ts >= r.ts).
+-- Exercises a right input that is ten times larger than the left input.
+WITH left_input AS (
+    SELECT value AS ts, value AS payload FROM range(100000)
+),
+right_input AS (
+    SELECT value AS ts, value AS payload FROM range(1000000)
+)
+SELECT l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts);

benchmarks/sql_benchmarks/asof_join/benchmarks/q06.benchmark

@@ -0,0 +1,25 @@
+name Q06
+group asof_join
+
+expect_plan AsOfJoinExec
+
+run
+-- Left and right: 1M rows each; one equality group contains 95% of rows.
+-- Inputs are not ordered by (key, ts), so required ordering is included.
+-- Matches the latest right row at or before each left timestamp (l.ts >= r.ts).
+WITH left_input AS (
+    SELECT CASE WHEN value % 100 < 95 THEN 0 ELSE value % 16 + 1 END AS key,
+           value / 100 + 1 AS ts,
+           value AS payload
+    FROM range(1000000)
+),
+right_input AS (
+    SELECT CASE WHEN value % 100 < 95 THEN 0 ELSE value % 16 + 1 END AS key,
+           value / 100 AS ts,
+           value AS payload
+    FROM range(1000000)
+)
+SELECT l.key, l.ts, l.payload, r.payload AS right_payload
+FROM left_input l
+ASOF JOIN right_input r MATCH_CONDITION (l.ts >= r.ts)
+ON l.key = r.key;

benchmarks/sql_benchmarks/asof_join/benchmarks/q07.benchmark

@@ -0,0 +1,25 @@
+name Q07
+group asof_join
+
+load sql_benchmarks/asof_join/init/load_q07.sql
+
+assert II
+SELECT
+    (SELECT count(*) FROM asof_left_sorted) AS left_rows,
+    (SELECT count(*) FROM asof_right_sorted) AS right_rows
+----
+100000|100000
+
+expect_plan AsOfJoinExec
+
+run
+-- Left and right: 100K rows each; 10K balanced equality groups; integer payloads.
+-- Both Parquet inputs declare (group_key, ts) ordering, so the measured query does not
+-- include the input sorts required by the other keyed workloads.
+-- Matches the latest right row at or before each left timestamp (l.ts >= r.ts).
+SELECT l.group_key, l.ts, l.payload, r.payload AS right_payload
+FROM asof_left_sorted l
+ASOF JOIN asof_right_sorted r MATCH_CONDITION (l.ts >= r.ts)
+ON l.group_key = r.group_key;
+
+cleanup sql_benchmarks/asof_join/init/cleanup_q07.sql

benchmarks/sql_benchmarks/asof_join/init/cleanup_q07.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS asof_left_sorted;
+DROP TABLE IF EXISTS asof_right_sorted;

benchmarks/sql_benchmarks/asof_join/init/load_q07.sql

@@ -0,0 +1,17 @@
+CREATE EXTERNAL TABLE asof_left_sorted (
+    group_key BIGINT,
+    ts BIGINT,
+    payload BIGINT
+)
+STORED AS PARQUET
+LOCATION '${DATA_DIR:-data}/asof_join/q07_left.parquet'
+WITH ORDER (group_key ASC NULLS FIRST, ts ASC NULLS FIRST);
+
+CREATE EXTERNAL TABLE asof_right_sorted (
+    group_key BIGINT,
+    ts BIGINT,
+    payload BIGINT
+)
+STORED AS PARQUET
+LOCATION '${DATA_DIR:-data}/asof_join/q07_right.parquet'
+WITH ORDER (group_key ASC NULLS FIRST, ts ASC NULLS FIRST);