bug: TPCH 18 query hangs

[Omega359]

Describe the bug

On my machine

OS: Fedora Linux 43 (KDE Plasma Desktop Edition) x86_64
Kernel: Linux 6.19.11-200.fc43.x86_64
CPU: AMD Ryzen AI 9 HX 370 (24) @ 5.16 GHz
GPU: AMD Radeon 890M Graphics [Integrated]
Memory: 11.96 GiB / 86.02 GiB (14%)
Swap: 0 B / 8.00 GiB (0%)

with updated rust:

$ rustup show
Default host: x86_64-unknown-linux-gnu
rustup home:  /home/bruce/.rustup

installed toolchains
--------------------
stable-x86_64-unknown-linux-gnu (default)
nightly-x86_64-unknown-linux-gnu
1.91.0-x86_64-unknown-linux-gnu
1.92.0-x86_64-unknown-linux-gnu
1.93.0-x86_64-unknown-linux-gnu
1.94.0-x86_64-unknown-linux-gnu (active)

active toolchain
----------------
name: 1.94.0-x86_64-unknown-linux-gnu
active because: overridden by '/home/bruce/dev/datafusion2/rust-toolchain.toml'
installed targets:
  x86_64-unknown-linux-gnu

Running against main:
cd benchmarks;./bench.sh data tpch;./bench.sh run tpch 18 will hang

$ ./bench.sh run tpch 18
***************************
DataFusion Benchmark Script
COMMAND: run
BENCHMARK: tpch
QUERY: 18
DATAFUSION_DIR: /home/bruce/dev/datafusion2/benchmarks/..
BRANCH_NAME: HEAD
DATA_DIR: /home/bruce/dev/datafusion2/benchmarks/data
RESULTS_DIR: /home/bruce/dev/datafusion2/benchmarks/results/HEAD
CARGO_COMMAND: cargo run --release
PREFER_HASH_JOIN: true
SIMULATE_LATENCY: false
***************************
RESULTS_FILE: /home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json
Running tpch benchmark...
+ cargo run --release --bin dfbench -- tpch --iterations 5 --path /home/bruce/dev/datafusion2/benchmarks/data/tpch_sf1 --prefer_hash_join true --format parquet -o /home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json --query 18
    Finished `release` profile [optimized] target(s) in 0.11s
     Running `/home/bruce/dev/datafusion2/target/release/dfbench tpch --iterations 5 --path /home/bruce/dev/datafusion2/benchmarks/data/tpch_sf1 --prefer_hash_join true --format parquet -o /home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json --query 18`
Running benchmarks with the following options: RunOpt { query: Some(18), common: CommonOpt { iterations: 5, partitions: None, batch_size: None, mem_pool_type: "fair", memory_limit: None, sort_spill_reservation_bytes: None, debug: false, simulate_latency: false }, path: "/home/bruce/dev/datafusion2/benchmarks/data/tpch_sf1", file_format: "parquet", mem_table: false, output_path: Some("/home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json"), disable_statistics: false, prefer_hash_join: true, enable_piecewise_merge_join: false, sorted: false, hash_join_buffering_capacity: 0 }

git bisect points to this commit as the cause. Running the test at the commit just prior to that one succeeds. Running it at that commit fails.

If prefer_hash_join is disabled the query will run as expected:

PREFER_HASH_JOIN=false ./bench.sh run tpch 18
***************************
DataFusion Benchmark Script
COMMAND: run
BENCHMARK: tpch
QUERY: 18
DATAFUSION_DIR: /home/bruce/dev/datafusion2/benchmarks/..
BRANCH_NAME: HEAD
DATA_DIR: /home/bruce/dev/datafusion2/benchmarks/data
RESULTS_DIR: /home/bruce/dev/datafusion2/benchmarks/results/HEAD
CARGO_COMMAND: cargo run --release
PREFER_HASH_JOIN: false
SIMULATE_LATENCY: false
***************************
RESULTS_FILE: /home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json
Running tpch benchmark...
+ cargo run --release --bin dfbench -- tpch --iterations 5 --path /home/bruce/dev/datafusion2/benchmarks/data/tpch_sf1 --prefer_hash_join false --format parquet -o /home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json --query 18
    Finished `release` profile [optimized] target(s) in 0.15s
     Running `/home/bruce/dev/datafusion2/target/release/dfbench tpch --iterations 5 --path /home/bruce/dev/datafusion2/benchmarks/data/tpch_sf1 --prefer_hash_join false --format parquet -o /home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json --query 18`
Running benchmarks with the following options: RunOpt { query: Some(18), common: CommonOpt { iterations: 5, partitions: None, batch_size: None, mem_pool_type: "fair", memory_limit: None, sort_spill_reservation_bytes: None, debug: false, simulate_latency: false }, path: "/home/bruce/dev/datafusion2/benchmarks/data/tpch_sf1", file_format: "parquet", mem_table: false, output_path: Some("/home/bruce/dev/datafusion2/benchmarks/results/HEAD/tpch_sf1.json"), disable_statistics: false, prefer_hash_join: false, enable_piecewise_merge_join: false, sorted: false, hash_join_buffering_capacity: 0 }
Query 18 iteration 0 took 206.5 ms and returned 57 rows
Query 18 iteration 1 took 190.0 ms and returned 57 rows
Query 18 iteration 2 took 188.5 ms and returned 57 rows
Query 18 iteration 3 took 185.0 ms and returned 57 rows
Query 18 iteration 4 took 192.1 ms and returned 57 rows
Query 18 avg time: 192.42 ms
+ set +x
Done

To Reproduce

This seems to be machine/OS specific. I've been unable to reproduce on other machines.

Expected behavior

No response

Additional context

No response

[adriangb]

@kosiew could you take a look?

maybe manually setting target partitions could help reproduce?

[RatulDawar]

Hey @adriangb can I debug this ? Would also get some context around datafusion, really interested in this project.

[adriangb]

Sure it can't help to get more help debugging 😃

[RatulDawar]

take

[adriangb]

Since you're new to this codebase I'd focus on reproducing the issue with a MRE that isn't OS/machine specific; don't worry about the fix too much for now.

[Omega359]

On my machine (12 cores, 24 with smt) running with DATAFUSION_EXECUTION_TARGET_PARTITIONS=11 works but DATAFUSION_EXECUTION_TARGET_PARTITIONS=12 (or more) hangs. That is with prefer_hash_join=true

[RatulDawar]

@Omega359 for me it's working always with prefer_hash_join=false and with prefer_hash_join=true it's failing for target_parititions > 12.

Also I see that a working instance with target partitions 4 gives a 400% CPU utiliztion vs a hanged instance gives 0% CPU utilization.

Probably suggesting that there a thread waiting/deadlock issue. Updating my findings further.

[RatulDawar]

This can be minimally reproduced using this (essentially the same as the bench query)

cargo build --release -p datafusion-cli

## use the cli 
DATAFUSION_EXECUTION_TARGET_PARTITIONS=24 ./target/release/datafusion-cli 
-- Register TPC-H Tables
CREATE EXTERNAL TABLE lineitem STORED AS PARQUET LOCATION 'benchmarks/data/tpch_sf1/lineitem';
CREATE EXTERNAL TABLE orders STORED AS PARQUET LOCATION 'benchmarks/data/tpch_sf1/orders';
CREATE EXTERNAL TABLE customer STORED AS PARQUET LOCATION 'benchmarks/data/tpch_sf1/customer';

-- TPC-H Query 18 (The hanging query)
select
    c_name,
    c_custkey,
    o_orderkey,
    o_orderdate,
    o_totalprice,
    sum(l_quantity)
from
    customer,
    orders,
    lineitem
where
        o_orderkey in (
        select
            l_orderkey
        from
            lineitem
        group by
            l_orderkey having
                sum(l_quantity) > 300
    )
  and c_custkey = o_custkey
  and o_orderkey = l_orderkey
group by
    c_name,
    c_custkey,
    o_orderkey,
    o_orderdate,
    o_totalprice
order by
    o_totalprice desc,
    o_orderdate
limit 100;

[RatulDawar]

@adriangb @Omega359 So I want able to find the issue here, the commit introduced a short circuit optimization where suppose if after the build process of the local partition there are zero rows, it doesn't need to run the probe state (as obviously there won't be any matches).
So this optimization basically marks the state from build -> completed.

Problem arises when we are using dynamic filtering since dynamic filter is shared across partitions (since it's used to filter the probe side table) we use synchronization across parititons so that before passing the filter to probe side we have completed the filter collection from all the build side partitions.

Now when we significantly increase the target partitions count some partitions get zero rows allocated, which leads to the shot-circuit getting triggerd marking state of the empty partitions to completed.

Now these empty partitions never report the dynamic filters to the shared accumlator, which is waiting for the status to be reported by all the parititons. Leading to indefinite weight.

What I propose as a fix here is that for empty partitons we still report the filters to the shared accumlator and then mark it as completed. This would stop this indefinite weight situation.
If you guys are aligned I will raise a PR with this fix.

TL;DR:
The short-circuit optimization skips probe execution for empty partitions, but those partitions never report their dynamic filters. Since the shared accumulator waits for all partitions, this causes an indefinite wait. Fix: even empty partitions should report (empty) filters before marking themselves as completed.

[adriangb]

That makes complete sense to me, great investigation @RatulDawar ! Do you want to work on the fix itself?

[RatulDawar]

Yes, would love to ! Sharing the fix