TPCH queries hang in benchmarks

[gabotechs]

While playing with Add AWS CDK-based benchmarking environment, I'm seeing that queries are likely to hang forever while running the benchmarks, printing Query still running... in a loop with nothing happening.

I wonder if there is an issue that's not captured by the current tests because of the fact they use local files rather than S3 ones.

I attempted printing the StageKeys in the ttl_map that stores the state of each query to see if there's something wrong there, and this is what I saw:

output log

[2025-11-18T12:40:46Z INFO  worker] Executing query...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:40:51Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:40:56Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:01Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:06Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:11Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:16Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:21Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:26Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:31Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:36Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:41Z INFO  worker] Query still running...
[StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 8, task_number: 0 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 4, task_number: 1 }, StageKey { query_id: b"\xb2\x7f\xe9\xa6\xd6\xc2O\x86\x82\xd1\x96\x03\\\x04\x1e0", stage_id: 3, task_number: 1 }]
[2025-11-18T12:41:46Z INFO  worker] Query still running...
[]
[2025-11-18T12:41:51Z INFO  worker] Query still running...
[]
[2025-11-18T12:41:56Z INFO  worker] Query still running...
[]
[2025-11-18T12:42:01Z INFO  worker] Query still running...
[]
[2025-11-18T12:42:06Z INFO  worker] Query still running...
[]
[2025-11-18T12:42:11Z INFO  worker] Query still running...
...
...
...
it goes like this forever

It looks like there's a deadlock somewhere in the code that completely stalls the query.

It seems to be triggered by some queries in particular, for example, TPCH query 7. It can be reproduced ~50% of the times running:

npm run datafusion-bench  -- --sf 10 --files-per-task 4 --query 7

With the remote benchmarks

[gabotechs]

It consistently happens on Stage 5 in this plan:

    ┌───── DistributedExec ── Tasks: t0:[p0]
    │ SortPreservingMergeExec: [supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC NULLS LAST, l_year@2 ASC NULLS LAST]
    │   [Stage 8] => NetworkCoalesceExec: output_partitions=8, input_tasks=2
    └──────────────────────────────────────────────────
      ┌───── Stage 8 ── Tasks: t0:[p0..p3] t1:[p0..p3]
      │ SortExec: expr=[supp_nation@0 ASC NULLS LAST, cust_nation@1 ASC NULLS LAST, l_year@2 ASC NULLS LAST], preserve_partitioning=[true]
      │   ProjectionExec: expr=[supp_nation@0 as supp_nation, cust_nation@1 as cust_nation, l_year@2 as l_year, sum(shipping.volume)@3 as revenue]
      │     AggregateExec: mode=FinalPartitioned, gby=[supp_nation@0 as supp_nation, cust_nation@1 as cust_nation, l_year@2 as l_year], aggr=[sum(shipping.volume)]
      │       CoalesceBatchesExec: target_batch_size=8192
      │         [Stage 7] => NetworkShuffleExec: output_partitions=4, input_tasks=1
      └──────────────────────────────────────────────────
        ┌───── Stage 7 ── Tasks: t0:[p0..p7]
        │ RepartitionExec: partitioning=Hash([supp_nation@0, cust_nation@1, l_year@2], 8), input_partitions=4
        │   AggregateExec: mode=Partial, gby=[supp_nation@0 as supp_nation, cust_nation@1 as cust_nation, l_year@2 as l_year], aggr=[sum(shipping.volume)]
        │     ProjectionExec: expr=[n_name@4 as supp_nation, n_name@0 as cust_nation, date_part(YEAR, l_shipdate@3) as l_year, l_extendedprice@1 * (Some(1),20,0 - l_discount@2) as volume]
        │       CoalesceBatchesExec: target_batch_size=8192
        │         HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, c_nationkey@3)], filter=n_name@0 = FRANCE AND n_name@1 = GERMANY OR n_name@0 = GERMANY AND n_name@1 = FRANCE, projection=[n_name@1, l_extendedprice@2, l_discount@3, l_shipdate@4, n_name@6]
        │           CoalescePartitionsExec
        │             CoalesceBatchesExec: target_batch_size=8192
        │               FilterExec: n_name@1 = GERMANY OR n_name@1 = FRANCE
        │                 RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
        │                   DataSourceExec: file_groups={1 group: [[tpch_sf10/nation/part-0.parquet]]}, projection=[n_nationkey, n_name], file_type=parquet, predicate=n_name@1 = GERMANY OR n_name@1 = FRANCE, pruning_predicate=n_name_null_count@2 != row_count@3 AND n_name_min@0 <= GERMANY AND GERMANY <= n_name_max@1 OR n_name_null_count@2 != row_count@3 AND n_name_min@0 <= FRANCE AND FRANCE <= n_name_max@1, required_guarantees=[n_name in (FRANCE, GERMANY)]
        │           ProjectionExec: expr=[l_extendedprice@1 as l_extendedprice, l_discount@2 as l_discount, l_shipdate@3 as l_shipdate, c_nationkey@4 as c_nationkey, n_name@0 as n_name]
        │             CoalesceBatchesExec: target_batch_size=8192
        │               HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@0)], projection=[n_name@1, l_extendedprice@3, l_discount@4, l_shipdate@5, c_nationkey@6]
        │                 CoalescePartitionsExec
        │                   CoalesceBatchesExec: target_batch_size=8192
        │                     FilterExec: n_name@1 = FRANCE OR n_name@1 = GERMANY
        │                       RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
        │                         DataSourceExec: file_groups={1 group: [[tpch_sf10/nation/part-0.parquet]]}, projection=[n_nationkey, n_name], file_type=parquet, predicate=n_name@1 = FRANCE OR n_name@1 = GERMANY, pruning_predicate=n_name_null_count@2 != row_count@3 AND n_name_min@0 <= FRANCE AND FRANCE <= n_name_max@1 OR n_name_null_count@2 != row_count@3 AND n_name_min@0 <= GERMANY AND GERMANY <= n_name_max@1, required_guarantees=[n_name in (FRANCE, GERMANY)]
        │                 ProjectionExec: expr=[s_nationkey@1 as s_nationkey, l_extendedprice@2 as l_extendedprice, l_discount@3 as l_discount, l_shipdate@4 as l_shipdate, c_nationkey@0 as c_nationkey]
        │                   CoalesceBatchesExec: target_batch_size=8192
        │                     HashJoinExec: mode=Partitioned, join_type=Inner, on=[(c_custkey@0, o_custkey@4)], projection=[c_nationkey@1, s_nationkey@2, l_extendedprice@3, l_discount@4, l_shipdate@5]
        │                       CoalesceBatchesExec: target_batch_size=8192
        │                         [Stage 1] => NetworkShuffleExec: output_partitions=4, input_tasks=4
        │                       CoalesceBatchesExec: target_batch_size=8192
        │                         [Stage 6] => NetworkShuffleExec: output_partitions=4, input_tasks=4
        └──────────────────────────────────────────────────
          ┌───── Stage 1 ── Tasks: t0:[p0..p3] t1:[p0..p3] t2:[p0..p3] t3:[p0..p3]
          │ RepartitionExec: partitioning=Hash([c_custkey@0], 4), input_partitions=1
          │   PartitionIsolatorExec: t0:[p0,__,__,__] t1:[__,p0,__,__] t2:[__,__,p0,__] t3:[__,__,__,p0] 
          │     DataSourceExec: file_groups={4 groups: [[tpch_sf10/customer/part-0.parquet:0..5842037, tpch_sf10/customer/part-1.parquet:0..5811593, tpch_sf10/customer/part-10.parquet:0..5787611, tpch_sf10/customer/part-11.parquet:0..5751851], [tpch_sf10/customer/part-11.parquet:5751851..5774089, tpch_sf10/customer/part-12.parquet:0..5771821, tpch_sf10/customer/part-13.parquet:0..5769125, tpch_sf10/customer/part-14.parquet:0..5772856, tpch_sf10/customer/part-15.parquet:0..5771904, ...], [tpch_sf10/customer/part-2.parquet:85148..5806970, tpch_sf10/customer/part-3.parquet:0..5813866, tpch_sf10/customer/part-4.parquet:0..5815060, tpch_sf10/customer/part-5.parquet:0..5811582, tpch_sf10/customer/part-6.parquet:0..30762], [tpch_sf10/customer/part-6.parquet:30762..5808127, tpch_sf10/customer/part-7.parquet:0..5802238, tpch_sf10/customer/part-8.parquet:0..5805923, tpch_sf10/customer/part-9.parquet:0..5807565]]}, projection=[c_custkey, c_nationkey], file_type=parquet, predicate=DynamicFilterPhysicalExpr [ true ]
          └──────────────────────────────────────────────────
          ┌───── Stage 6 ── Tasks: t0:[p0..p3] t1:[p0..p3] t2:[p0..p3] t3:[p0..p3]
          │ RepartitionExec: partitioning=Hash([o_custkey@4], 4), input_partitions=4
          │   CoalesceBatchesExec: target_batch_size=8192
          │     HashJoinExec: mode=Partitioned, join_type=Inner, on=[(l_orderkey@1, o_orderkey@0)], projection=[s_nationkey@0, l_extendedprice@2, l_discount@3, l_shipdate@4, o_custkey@6]
          │       CoalesceBatchesExec: target_batch_size=8192
          │         [Stage 4] => NetworkShuffleExec: output_partitions=4, input_tasks=4
          │       CoalesceBatchesExec: target_batch_size=8192
          │         [Stage 5] => NetworkShuffleExec: output_partitions=4, input_tasks=4
          └──────────────────────────────────────────────────
            ┌───── Stage 4 ── Tasks: t0:[p0..p15] t1:[p0..p15] t2:[p0..p15] t3:[p0..p15]
            │ RepartitionExec: partitioning=Hash([l_orderkey@1], 16), input_partitions=4
            │   CoalesceBatchesExec: target_batch_size=8192
            │     HashJoinExec: mode=Partitioned, join_type=Inner, on=[()], projection=[s_nationkey@1, l_orderkey@2, l_extendedprice@4, l_discount@5, l_shipdate@6]
            │       CoalesceBatchesExec: target_batch_size=8192
            │         [Stage 2] => NetworkShuffleExec: output_partitions=4, input_tasks=4
            │       CoalesceBatchesExec: target_batch_size=8192
            │         [Stage 3] => NetworkShuffleExec: output_partitions=4, input_tasks=4
            └──────────────────────────────────────────────────
              ┌───── Stage 2 ── Tasks: t0:[p0..p15] t1:[p0..p15] t2:[p0..p15] t3:[p0..p15]
              │ RepartitionExec: partitioning=Hash([s_suppkey@0], 16), input_partitions=1
              │   PartitionIsolatorExec: t0:[p0,__,__,__] t1:[__,p0,__,__] t2:[__,__,p0,__] t3:[__,__,__,p0] 
              │     DataSourceExec: file_groups={4 groups: [[tpch_sf10/supplier/part-0.parquet, tpch_sf10/supplier/part-1.parquet, tpch_sf10/supplier/part-10.parquet, tpch_sf10/supplier/part-11.parquet], [tpch_sf10/supplier/part-12.parquet, tpch_sf10/supplier/part-13.parquet, tpch_sf10/supplier/part-14.parquet, tpch_sf10/supplier/part-15.parquet], [tpch_sf10/supplier/part-2.parquet, tpch_sf10/supplier/part-3.parquet, tpch_sf10/supplier/part-4.parquet, tpch_sf10/supplier/part-5.parquet], [tpch_sf10/supplier/part-6.parquet, tpch_sf10/supplier/part-7.parquet, tpch_sf10/supplier/part-8.parquet, tpch_sf10/supplier/part-9.parquet]]}, projection=[s_suppkey, s_nationkey], file_type=parquet, predicate=DynamicFilterPhysicalExpr [ true ]
              └──────────────────────────────────────────────────
              ┌───── Stage 3 ── Tasks: t0:[p0..p15] t1:[p0..p15] t2:[p0..p15] t3:[p0..p15]
              │ RepartitionExec: partitioning=Hash([l_suppkey@1], 16), input_partitions=1
              │   CoalesceBatchesExec: target_batch_size=8192
              │     FilterExec: l_shipdate@4 >= 1995-01-01 AND l_shipdate@4 <= 1996-12-31
              │       PartitionIsolatorExec: t0:[p0,__,__,__] t1:[__,p0,__,__] t2:[__,__,p0,__] t3:[__,__,__,p0] 
              │         DataSourceExec: file_groups={4 groups: [[tpch_sf10/lineitem/part-0.parquet:0..104241221, tpch_sf10/lineitem/part-1.parquet:0..104028426, tpch_sf10/lineitem/part-10.parquet:0..103284844, tpch_sf10/lineitem/part-11.parquet:0..102113162], [tpch_sf10/lineitem/part-11.parquet:102113162..103237063, tpch_sf10/lineitem/part-12.parquet:0..103274897, tpch_sf10/lineitem/part-13.parquet:0..103278373, tpch_sf10/lineitem/part-14.parquet:0..103288811, tpch_sf10/lineitem/part-15.parquet:0..102701671], [tpch_sf10/lineitem/part-15.parquet:102701671..103299688, tpch_sf10/lineitem/part-2.parquet:0..103757802, tpch_sf10/lineitem/part-3.parquet:0..103260147, tpch_sf10/lineitem/part-4.parquet:0..103266335, tpch_sf10/lineitem/part-5.parquet:0..102785352], [tpch_sf10/lineitem/part-5.parquet:102785352..103281610, tpch_sf10/lineitem/part-6.parquet:0..103307501, tpch_sf10/lineitem/part-7.parquet:0..103302422, tpch_sf10/lineitem/part-8.parquet:0..103309760, tpch_sf10/lineitem/part-9.parquet:0..103251710]]}, projection=[l_orderkey, l_suppkey, l_extendedprice, l_discount, l_shipdate], file_type=parquet, predicate=l_shipdate@4 >= 1995-01-01 AND l_shipdate@4 <= 1996-12-31 AND DynamicFilterPhysicalExpr [ true ], pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_max@0 >= 1995-01-01 AND l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@3 <= 1996-12-31, required_guarantees=[]
              └──────────────────────────────────────────────────
            ┌───── Stage 5 ── Tasks: t0:[p0..p15] t1:[p0..p15] t2:[p0..p15] t3:[p0..p15]
            │ RepartitionExec: partitioning=Hash([o_orderkey@0], 16), input_partitions=1
            │   PartitionIsolatorExec: t0:[p0,__,__,__] t1:[__,p0,__,__] t2:[__,__,p0,__] t3:[__,__,__,p0] 
            │     DataSourceExec: file_groups={4 groups: [[tpch_sf10/orders/part-0.parquet:0..25621343, tpch_sf10/orders/part-1.parquet:0..25565034, tpch_sf10/orders/part-10.parquet:0..25354296, tpch_sf10/orders/part-11.parquet:0..25021731], [tpch_sf10/orders/part-11.parquet:25021731..25357002, tpch_sf10/orders/part-12.parquet:0..25353629, tpch_sf10/orders/part-13.parquet:0..25353334, tpch_sf10/orders/part-14.parquet:0..25352147, tpch_sf10/orders/part-15.parquet:0..25168023], [tpch_sf10/orders/part-15.parquet:25168023..25354984, tpch_sf10/orders/part-2.parquet:0..25489977, tpch_sf10/orders/part-3.parquet:0..25356374, tpch_sf10/orders/part-4.parquet:0..25351435, tpch_sf10/orders/part-5.parquet:0..25177657], [tpch_sf10/orders/part-5.parquet:25177657..25347961, tpch_sf10/orders/part-6.parquet:0..25346334, tpch_sf10/orders/part-7.parquet:0..25346005, tpch_sf10/orders/part-8.parquet:0..25347754, tpch_sf10/orders/part-9.parquet:0..25352005]]}, projection=[o_orderkey, o_custkey], file_type=parquet, predicate=DynamicFilterPhysicalExpr [ true ] AND DynamicFilterPhysicalExpr [ true ]
            └──────────────────────────────────────────────────

It's like something in Stage 5 deadlocks and hangs forever

[gabotechs]

@adriangb or @marc-pydantic do you happen to have noticed something similar in your setup? I'm unable to replicate this in any other environment apart from the EC2-based one in AWS

[gabotechs]

Closed by #240