Resolve lost wakeup in SpillPoolReader with multiple concurrent SpillPoolWriters by pepijnve · Pull Request #23522 · apache/datafusion

pepijnve · 2026-07-13T00:37:02Z

Which issue does this PR close?

Closes SMJ + aggregate finish with one partition but stuck forever when number of partitions is big #23447

Rationale for this change

When multiple SpillPoolWriter clones concurrently push batches to the same channel, more than one non-finished SpillFile can be in flight. This happens because each SpillPoolWriter clone takes the current_write_file at the start of push_batch and puts it back when it's done. When multiple push_batch calls happen concurrently, only the first one will be able to take the current_write_file and the others will all create their own new spill file. Which one gets put back for subsequent use is a race condition.

If this occurred and the writers are all dropped before rotation happens in, multiple files in the files deque will be have writer_finished == false. The last writer drop logic in SpillPoolWriter::drop only finishes whatever file is the current_write_file as finished.

This can lead to a stalled situation when SpillPoolFile::poll_next catches up with the writer and returns Pending because writer_finished == false. A waker for the file is registered, but since the last writer drop logic only finishes and wakes whatever happens to be current_write_file, which may not be the current read file, the waker may end up never being notified.

There is a secondary waker that is registered on the spill pool itself, but due to fine grained locking, it is possible for the wake call in the last writer drop logic to be called before the waker registration.

What changes are included in this PR?

Add support for tracking multiple unfinished write files
Close all unfinished write files when the last writer is dropped
Removed writer_dropped field which was an unnecessary denormalisation of active_writer_count == 0

An additional benefit of tracking all unfinished write files is that excessive creation of tiny spill files is avoided when many writers are pushing batches concurrently.

Are these changes tested?

Reproduction case from linked issue was used to confirm fix

Are there any user-facing changes?

No

SemyonSinchenko · 2026-07-13T08:54:46Z

An additional benefit of tracking all unfinished write files is that excessive creation of tiny spill files is avoided when many writers are pushing batches concurrently.

That is interesting because I constantly facing the "Too many open files" when there is a heavy spill. I did not know it is a problem and I was thinking it is OK, just used ulimit -n 524288.

jayzhan211 · 2026-07-14T12:35:05Z

+        if !shared.current_write_files.is_empty() {
+            // Copy and clear `current_write_files` so we can release shared lock before locking files
+            let files = shared.current_write_files.clone();
+            shared.current_write_files.clear();


Do we really need to release shared lock before locking files

Does it make sense to take file out from current_write_files consecutively then we could avoid clone?

Do we really need to release shared lock before locking files

I'm not entirely sure. My assumption was that this was already carefully tuned, so I didn't want to change the lock nesting. If I understood it correctly, the intention is to avoid IO while the shared lock is held. InProgressSpilFile::finish can do IO so the shared lock is released before calling finish.

Does it make sense to take file out from current_write_files consecutively then we could avoid clone?

The clone/clear is actually kind of pointless. Might as well just mem::take the entire VecDequeue here. I've made that change.

I'm not entirely sure. My assumption was that this was already carefully tuned, so I didn't want to change the lock nesting. If I understood it correctly, the intention is to avoid IO while the shared lock is held. InProgressSpilFile::finish can do IO so the shared lock is released before calling finish.

we also need to be sure we don't have a deadlock due to lock inversion (aka try to take the file/shared locks in different orders across different code paths)

alamb · 2026-07-15T12:58:44Z

+            file_shared.writer_finished = true;
+            // Wake reader waiting on this file (it's now finished)
+            file_shared.wake();
+            // Don't put back current_write_file - let it rotate


Suggested change

// Don't put back current_write_file - let it rotate

// Don't put back current_write_files - let it rotate

Singular is actually correct here, but I'll change the wording a bit to clarify what's being done.

alamb

This looks good to me -- thank you @pepijnve and @jayzhan211 (BTW 👋 it is great to see you again). I think adding some more comments explaining the expected state of the various queues would help make this code easier to understand in the future, but doesn't have to happen as part of this PR

cc @adriangb who added I think added SpillPoolShared in the first place

#18207

cc @xanderbailey who added some of this share pool machnery

#20672

alamb · 2026-07-15T13:04:01Z

 use super::spill_manager::SpillManager;

 /// Shared state between the writer and readers of a spill pool.
 /// This contains the queue of files and coordination state.


I this these comments would be nice to update to reflect the new design

Specifically note that the current_write_files is a list of files that are currently open but not being acively written to. When a file is being actively written to, active writer count is incremented by one and an entry is removed from current_write_files

I'll do another pass over the code and update the comments. I didn't pay sufficient attention to those.

alamb · 2026-07-15T13:05:32Z


            // No files in queue - check if writer is done
-            if shared.writer_dropped {
+            if shared.active_writer_count == 0 {


I wonder (maybe as a follow on PR) if adding some sort of RAAI guard for updating active_writer_count would make the code les error prone.

Or maybe even having a separate list of files being actively written (rather than just a count 🤔 )

The SpillPoolWriter plays the role of RAAI guard itself, doesn't it? It's incremented when a SpillPoolWriter is cloned, and decremented on drop.

The name active_writer_count is a bit misleading. It does not reflect the number of writers currently writing a batch. It reflects the number of writers that still exist.

alamb · 2026-07-15T13:07:08Z

+        if !shared.current_write_files.is_empty() {
+            // Copy and clear `current_write_files` so we can release shared lock before locking files
+            let files = shared.current_write_files.clone();
+            shared.current_write_files.clear();


I'm not entirely sure. My assumption was that this was already carefully tuned, so I didn't want to change the lock nesting. If I understood it correctly, the intention is to avoid IO while the shared lock is held. InProgressSpilFile::finish can do IO so the shared lock is released before calling finish.

we also need to be sure we don't have a deadlock due to lock inversion (aka try to take the file/shared locks in different orders across different code paths)

pepijnve · 2026-07-15T15:13:22Z

I've updated the documentation and code comments for correctness. I've deemphasised FIFO semantics quite a bit. The existing documentation stated that this was always guaranteed, but that was not actually the case. Prior to this PR the following chain of events could happen:

Writer A starts a batch push of B1, takes current_write_file F1
Writer B starts a batch push B2, current_write_file is None so it starts a new file F2
Writer B finishes its batch push, and sets current_write_file to F2
Writer A finished its batch push, and sets current_write_file to F1
Writer B pushes another batch B3 which is written current_write_file which is F1

The reader will observe batches in the order B1, B3, B2.

This MR assumes that we can take this even further, since the FIFO guarantee already doesn't hold when multiple writers are present. Would be good to get some feedback from @adriangb indeed since the non-FIFO behaviour was already present in the first commit of this code unless I'm completely misreading it.

I'm going to try to write a test that demonstrates the behaviour.

adriangb · 2026-07-15T15:31:26Z

I'm worried about dropping the FIFO semantics (even if they were broken).
Consider a plan like Projection(inner=Repartition(inner=Sort, preserve_order=true))): if RepartitionExec does not actually preserve order under spilling then the sort order will be lost and the results will be wrong.

pepijnve · 2026-07-15T15:50:16Z

I'm worried about dropping the FIFO semantics (even if they were broken). Consider a plan like Projection(inner=Repartition(inner=Sort, preserve_order=true))): if RepartitionExec does not actually preserve order under spilling then the sort order will be lost and the results will be wrong.

I'm probably missing a bit too much context. Could you help me define the exact order guarantee that's being promised by the spill pool? Single writer remains FIFO, but in the presence of multiple concurrent writers, what order can you actually guarantee for the reader? The best you can do I think is guarantee that the relative order of the batches per writer is retained. Is that sufficient?

pepijnve · 2026-07-15T16:35:27Z

BTW, I had based myself a bit on this comment in RepartitionExecState

// Create spill channels based on mode:
// - preserve_order: one spill channel per (input, output) pair for proper FIFO ordering
// - non-preserve-order: one shared spill channel per output partition since all inputs
//   share the same receiver

When preserve_order is requested, you get an SPSC channel per input and FIFO is guaranteed. When it is not requested, you get a single MPSC channel for all inputs and order is undefined. That aspect hasn't changed. What I've tried to clarify in the documentation is that the FIFO guarantee is only in the SPSC case, not the MPSC one.

adriangb · 2026-07-15T17:14:53Z

That makes sense, I was just checking that and came to the same conclusion. As long as SPSC preserves ordering / FIFO we are good 👍🏻

The spill pool's FIFO guarantee only holds for a single writer: with multiple concurrent `SpillPoolWriter` clones the reader can observe batches out of write order. That is fine for non-preserve-order `RepartitionExec` (the output is an unordered multiset), but for `preserve_order = true` the per-(input, output) stream feeds an order-sensitive `StreamingMerge`, so losing FIFO would silently produce wrong (unsorted) results. Previously the "single writer per ordered pool" invariant was upheld only by convention: one `preserve_order` bool drove two independent decisions (channel count vs. writer cloning) in two places, coupled only by comments. A future edit could break one without the other. Encode the invariant in the type system instead: - `channel()` now returns `SpillPoolWriter`, which is **not** `Clone`, so an ordered pool can only ever have one writer (enforced at compile time). It wraps the shared implementation and delegates `push_batch`. - `shared_channel()` returns the `Clone` `SharedSpillPoolWriter` for the multi-producer, per-writer-FIFO case. - `RepartitionExec` selects the topology in one place: `preserve_order` builds one dedicated ordered writer per input (moved, never cloned) via `PartitionSpillWriters::PerInput`; non-preserve builds one shared writer cloned across inputs via `PartitionSpillWriters::Shared`. Now feeding an ordered pool with a shared multi-producer writer simply does not compile. Adds `test_preserve_order_with_spill_file_rotation`, which forces a spill file per batch (`max_spill_file_size_bytes = 1`) and asserts each output partition stays sorted — exercising FIFO across file rotation. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WsjHVeKZtrSHugLwNFURjL

pepijnve · 2026-07-15T18:20:42Z

I've added a test_concurrent_writers test case that uses a 10 thread worker pool with 10 writers so that there's actual concurrency in the test. In contrast to test_concurrent_reader_writer there is no FIFO order assertion. Instead I'm asserting that no batches were lost.

Enforce single-writer spill pools for preserve_order via the type system

pepijnve · 2026-07-15T18:26:35Z

Thanks for the suggestion @adriangb I was thinking of doing something similar as well. I'll handle the merge conflict with main

# Conflicts: # datafusion/physical-plan/src/repartition/mod.rs

…e SpillWriter wrapper type

pepijnve · 2026-07-15T19:25:44Z

I flipped around the nesting of SpillPoolWriter and SharedSpillPoolWriter. You can then create a SpillPoolWriter via SharedSpillPoolWriter::new_writer. This has the benefit of not needing the SpillWriter wrapper in RepartitionExec. The tradeoff is that the writer type itself no longer strictly communicates the spsc vs mpsc and ordering characteristic of the channel. We still retain the guarantee from the compiler that channel is single producer and shared_channel can be multi producer.

I've edited the documentation a bit to be less verbose. Claude seemed kind of happy to repeat the fact that indeed SharedSpillPoolWriter is Clone over and over again.

jayzhan211 · 2026-07-16T14:15:24Z

+                let mut file_shared = file.lock();
+
+                // Finish the current writer if it exists
+                if let Some(mut writer) = file_shared.writer.take() {


alamb · 2026-07-16T18:13:55Z

@jayzhan211 is back! 🎉

alamb

This PR is looking really nice to me @pepijnve @adriangb and @jayzhan211

However, since the scope has significantly grown since I first reviewed the PR I don't think it would be a good idea to backport this change as is without some "bake time" on main first...

#22547

Perhaps we can apply the earlier (smaller) fix for 54?

alamb · 2026-07-16T18:16:57Z

+/// The set of spill-pool writers for a single output partition, before they are handed to the
+/// per-input tasks. The variant encodes the repartition mode so the wrong writer topology cannot
+/// be constructed for a given mode.
+enum PartitionSpillWriters {


this is much clearer

github-actions · 2026-07-16T18:28:30Z

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion-physical-plan v54.0.0 (current)
       Built [  35.668s] (current)
     Parsing datafusion-physical-plan v54.0.0 (current)
      Parsed [   0.147s] (current)
    Building datafusion-physical-plan v54.0.0 (baseline)
       Built [  35.298s] (baseline)
     Parsing datafusion-physical-plan v54.0.0 (baseline)
      Parsed [   0.150s] (baseline)
    Checking datafusion-physical-plan v54.0.0 -> v54.0.0 (no change; assume patch)
     Checked [   1.031s] 223 checks: 222 pass, 1 fail, 0 warn, 30 skip

--- failure function_marked_deprecated: function #[deprecated] added ---

Description:
A function is now #[deprecated]. Downstream crates will get a compiler warning when using this function.
        ref: https://doc.rust-lang.org/reference/attributes/diagnostics.html#the-deprecated-attribute
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.48.0/src/lints/function_marked_deprecated.ron

Failed in:
  function datafusion_physical_plan::spill::spill_pool::channel in /home/runner/work/datafusion/datafusion/datafusion/physical-plan/src/spill/spill_pool.rs:480

     Summary semver requires new minor version: 0 major and 1 minor checks failed
    Finished [  74.105s] datafusion-physical-plan

pepijnve · 2026-07-16T18:36:15Z

This PR is looking really nice to me @pepijnve @adriangb and @jayzhan211

However, since the scope has significantly grown since I first reviewed the PR I don't think it would be a good idea to backport this change as is without some "bake time" on main first...

Release DataFusion 54.1.0 (minor/patch) Release #22547

Perhaps we can apply the earlier (smaller) fix for 54?

I can go back and redo just the code changes and test additions on a back port branch. Makes sense to keep it as small as possible for a patch.

codecov-commenter · 2026-07-16T18:52:02Z

Codecov Report

❌ Patch coverage is 89.55224% with 21 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@95de385). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/repartition/mod.rs	86.41%	2 Missing and 9 partials ⚠️
datafusion/physical-plan/src/spill/spill_pool.rs	91.66%	6 Missing and 4 partials ⚠️

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #23522   +/-   ##
=======================================
  Coverage        ?   80.65%           
=======================================
  Files           ?     1086           
  Lines           ?   366430           
  Branches        ?   366430           
=======================================
  Hits            ?   295561           
  Misses          ?    53244           
  Partials        ?    17625

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

alamb · 2026-07-16T18:54:01Z

(I am resolving the CI failures now)

pepijnve · 2026-07-16T19:08:56Z

❌ Patch coverage is 89.55224% with 21 lines in your changes missing coverage. Please review.

Does every patch now require 100% coverage of changed lines?

alamb · 2026-07-16T19:37:35Z

Does every patch now require 100% coverage of changed lines?

No!

We are sorting through the coverage reports (and they are somewhat useless now b/c there isn't a base to compare against):

Code coverage report not working #23647

pepijnve · 2026-07-16T19:43:32Z

No!

That's a relief. In the meantime I've created the minimal backport PR version of this.

xanderbailey · 2026-07-16T21:52:01Z

This is a very nice improvement, thanks for working on this!

#23654) ## Which issue does this PR close? - part of #22547 54.x branch backport of fix for #23447 ## Rationale for this change See main PR #23522 ## What changes are included in this PR? See main PR #23522 ## Are these changes tested? Additional test cases added to verify existing behaviour Fix manually tested with reproduction code from #23447 ## Are there any user-facing changes? No

hareshkh · 2026-07-17T12:22:44Z

+    /// Maximum size in bytes before rotating to a new file.
+    /// Typically set from configuration `datafusion.execution.max_spill_file_size_bytes`.
+    max_file_size_bytes: usize,
+    /// Shared state with readers (includes current_write_file for coordination)


nit:

(includes current_write_file for coordination)

is stale

Since this branch has been merged already it would be best to make a PR with a fix

…PoolWriters (apache#23522) ## Which issue does this PR close? - Closes apache#23447 ## Rationale for this change When multiple `SpillPoolWriter` clones concurrently push batches to the same channel, more than one non-finished `SpillFile` can be in flight. This happens because each `SpillPoolWriter` clone takes the `current_write_file` at the start of `push_batch` and puts it back when it's done. When multiple `push_batch` calls happen concurrently, only the first one will be able to take the `current_write_file` and the others will all create their own new spill file. Which one gets put back for subsequent use is a race condition. If this occurred and the writers are all dropped before rotation happens in, multiple files in the `files` deque will be have `writer_finished == false`. The last writer drop logic in `SpillPoolWriter::drop` only finishes whatever file is the `current_write_file` as finished. This can lead to a stalled situation when `SpillPoolFile::poll_next` catches up with the writer and returns `Pending` because `writer_finished == false`. A waker for the file is registered, but since the last writer drop logic only finishes and wakes whatever happens to be `current_write_file`, which may not be the current read file, the waker may end up never being notified. There is a secondary waker that is registered on the spill pool itself, but due to fine grained locking, it is possible for the wake call in the last writer drop logic to be called before the waker registration. ## What changes are included in this PR? - Add support for tracking multiple unfinished write files - Close all unfinished write files when the last writer is dropped - Removed `writer_dropped` field which was an unnecessary denormalisation of `active_writer_count == 0` An additional benefit of tracking all unfinished write files is that excessive creation of tiny spill files is avoided when many writers are pushing batches concurrently. ## Are these changes tested? Reproduction case from linked issue was used to confirm fix ## Are there any user-facing changes? No --------- Co-authored-by: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

…PC-DS validation (#62) * fix(bench,deploy): SF10-distributed harness, CFN ShufflePartitions, bootstrap fixes - register-glue.sh: fix SIGPIPE false-empty on multi-file table listings (grep -q closed the pipe early under pipefail; multi-file SF10 tables were falsely reported empty and skipped) - weft-cluster.yaml + deploy-stack.sh: ShufflePartitions parameter wired to the weft:shuffle-partitions tag (was hardcoded to WorkerCount) - remeasure-distributed.sh: SF-agnostic (SF env, GLUE_DB_* defaults derive from SF, --sf passthrough) - run-spark-connect.py: --query-timeout (env WEFT_QUERY_TIMEOUT) wraps collect in a ThreadPoolExecutor; on timeout the query fails fast and the Spark session is recreated so a wedged query can't hang the suite - packer bootstrap.sh: fix enable_role_unit deadlock + spill device (the shipped AMI still has the buggy copy; rebake pending) - docs: SF100/SF10 topology, memory invariant, WEFT_SHUFFLE_PARTITIONS - SF10-DISTRIBUTED-RESUME.md: live-cluster handoff state * fix(execution): distributed stage lifecycle, spill, and correctness fixes (KAN-15..33) - KAN-15/17: remove 30s Flight timeout, narrow is_retryable, WEFT_STAGE_TIMEOUT_MS, cancel path end-to-end - KAN-18/19: stage cleanup on all exit paths, ShuffleInputGuard, WEFT_STAGE_OUTPUT_TTL_SECS - KAN-20: completed_ops cap/TTL, real ReleaseExecute - KAN-21/23: worker-global spill budget + incremental .segN appends - KAN-22: correlated scalar subquery decorrelation (Q2 class) - KAN-25: hash-join build guard WEFT_HASH_JOIN_MAX_BUILD_FRACTION - KAN-26: Q12/Q13 join chains, agg-over-agg - KAN-27: one-row broadcast via __WEFT_SCALAR_STAGE__ literal injection - KAN-28: empty results return typed zero rows - KAN-29: semi/anti/IN decorrelation, Q17 residual, Q15 derived scalar, strict whole-fact-gather floor - KAN-32: intermediate bucket dispatch at 16 partitions/2 workers, streaming producers, per-partition dispatch, AQE row-count action - KAN-33: Q13/Q16/Q20/Q22 distributed shapes Regression tests: auto_distribute_decorrelate, auto_distribute_semi_anti, auto_distribute_kan33, distributed_high_card_groupby, stage_timeout, stage_cancel, stage_cleanup, worker_join_memory_guard. * fix(loom): key catalog provider cache by shard context (KAN-35) The catalog bridge cached resolved table providers by table name only, but a provider bakes in the shard/replicate decision from first resolution. Per-query auto-broadcast classification flips table roles between queries, so stale providers served full scans where the plan assumed sharded (rows x worker count — TPC-H Q4 exactly 2x at SF10) or shards where the plan assumed replicated (~0.5x loss in Q5/Q7, 0.25x in Q9). Cache is now keyed by (table_name, is_replicated). Regression test: crates/weft-loom/tests/catalog_shard_context.rs (role-flip in both directions; fails with the old name-only key). * fix(connect): distributed strict-mode surface + result-path alignment (KAN-15..33 support) * fix(bench): SF-agnostic remeasure harness, query-timeout flag, Q11 fraction (KAN-30) - run-spark-connect: --query-timeout, --skip-worker-preflight, jsonl results - q11.sql: HAVING fraction now 0.0001 / __WEFT_SF__ (TPC-H spec) - weft-bench: tpch-distributed WEFT_TPCH_ONLY/DEBUG, SF-agnostic remeasure - docs/runtime-contract.md updated to match * fix(execution): distribute Q13/Q22 under size-based auto-broadcast (KAN-36, KAN-38) The KAN-33 shapes only handled both-tables-sharded configs. The real cluster's resolve_replicated_tables replicates all but the largest table, so Q13/Q22 run with customer replicated and orders sharded: - Q13: relaxed broadcast for null-extended-side-sharded LEFT/RIGHT joins when every aggregate reads only null-extended-side columns (NULL extension contributes nothing to partials); routes the agg-over-agg composition at sharded.len()==1. 3-stage plan. - Q22: classify_scalar_conjunct accepts fully-replicated scalar bodies (partial runs once via ExchangeMode::Forward); body_projection allowed over fully-replicated outer, hash-shuffled by shared key. 6-stage plan. Regression tests: crates/weft-execution/tests/auto_distribute_kan36.rs (5 tests; fail before/pass after; 2-worker Flight e2e == single-node). tpch-distributed sf0.01: 22/22 distributed-ok, 0 mismatch. * fix(execution): serve only same-schema shuffle entries; lock AVG precision (KAN-39, KAN-40) KAN-39: Worker::read_shuffle unioned every stage-cache entry matching a stage id and took the first entry's schema. Stage ids repeat across queries, and a client-timed-out query's producer could insert after the driver's cleanup raced past it, so the next same-id stage served mixed-schema bucket sets and register_batches failed with 'Mismatch between schema and batches'. Now the freshest entry declares the served schema, benign nullability/metadata drift is aligned (align_batch_schema), and genuinely stale entries are skipped with a warning; plus a registration-side alignment guard in run_stage. KAN-40: not an engine bug — distributed AVG is bit-for-bit identical to single-node (Spark-faithful DECIMAL(19,6) typing); added distributed_avg.rs to lock distributed==single-node at full precision. Regression tests: stale_entry_from_prior_query_is_not_served (fails before the fix), crates/weft-execution/tests/distributed_avg.rs. * fix(execution): fuse grouped IN-subquery with outer aggregate (KAN-37) Q18's generic semi/anti plan shuffled the full 3-way customer-orders- lineitem join output (~60M wide rows at SF10) across 16 partitions, then grouped it all to discard everything but 624 orders — exceeding the 600s stage timeout. The outer sum(l_quantity) grouped by a key set containing o_orderkey IS the IN-subquery's per-key aggregate, so the fact never needs to join the dims: new try_in_agg_semi_join shape emits 3 stages (partial per-key agg -> combine+HAVING carrying r0 -> co-located join + exact local GROUP BY with sum(r0) tracking join multiplicity), strictly guarded so non-matching shapes fall through to the generic path. Live SF10: Q18 42.0s cold / 34.5s warm (was >600s), 100 rows, row-for-row exact vs DuckDB golden incl. ORDER BY/LIMIT boundary. Local sf3: 20.6s. Regression tests in auto_distribute_semi_anti.rs (plan shape fail-before, distributed==single-node, decline guards). * fix(execution): exact combine for agg-over-agg through CTE boundary (KAN-44) TPC-DS Q54 at SF10 split one customer's revenue across two shuffle partitions and never recombined by the group key: the single-sharded peels_to_inner_aggregate divert only fired for the KAN-36 null-extended outer-join shape, so Q54's inner GROUP BY c_customer_sk (inside the my_revenue CTE) fell through to the flat broadcast path and executed per worker over its local shard. The divert is now unconditional at sharded.len()==1 — partial hashed by inner key -> combine (one row per inner group) -> re-shuffle by outer key -> exact outer aggregate — subsuming the KAN-36 special case. Non-provable shapes (expression outer group keys, grouping sets) are declined instead of mis-planned. Regression tests: crates/weft-execution/tests/auto_distribute_kan44.rs (3 tests, fail before/pass after, distributed==single-node with customer rows spanning both shards). tpch-distributed 22/22 0 mismatch; TPC-DS execute ratchet 95/95 0 mismatch. * fix(connect): stream typed empty batch for zero-row distributed SQL results (KAN-42) The distributed SQL branch of base_relation_batches returned an empty Vec<RecordBatch> verbatim — every other result path honors the KAN-28 zero-row contract. For a zero-row distributed result (e.g. TPC-DS Q17), stream_batches emitted only ResultComplete: no ArrowBatch, no schema, no error, and PySpark died at 'assert table is not None'. Now mirrors the local path: re-derive the schema and push one typed empty batch. Regression test: distributed_zero_row_result_still_streams_typed_empty_batch (crates/weft-connect/tests/distributed_pyspark.rs) — fails before, passes after; asserts >=1 ArrowBatch response on a real ExecutePlan stream. * fix(execution): materialize replicated-only branches once; dedup identical CTE branches (KAN-41) TPC-DS Q58/Q78 inlined replicated-only aggregate branches (catalog_sales/ web_sales arms) in the per-partition outer stage, recomputing a full replicated-fact scan+join+aggregate 16x at SF10 (>600s); Q11 planned 4 structurally identical self-joined CTE branch DAGs (17 stages, each fact scanned 4x). dag_splitter now materializes replicated-only aggregate branches as single Forward stages (computed once; volatile branches and branches whose expression subqueries read sharded tables stay inline) and dedups identical CTE branches by plan fingerprint below SubqueryAlias. Q11: 17 -> 5 stages. Q72/Q93 investigated: no plan defect (already join-at-shard + partial aggregate; fast live-faithful) — harness artifact locally (all-MemTable workers lose stats/pushdown), slow-but-correct. Regression tests: crates/weft-execution/tests/auto_distribute_kan41.rs (5 tests, shape tests fail before/pass after) + 4 dag_splitter unit tests. Gates green; tpch-distributed 22/22; tpcds ratchets 95/99 + 95/95 held. * fix(loom,execution): make sort-merge fallback opt-in; fail fast over wedging (KAN-45) Root cause (local repro + live evidence): DataFusion 54 deadlocks under the bounded FairSpillPool — a spilling operator stalls while its consumer stops draining a RepartitionExec channel, parking the stage at 0% CPU until WEFT_STAGE_TIMEOUT_MS kills it (upstream delta-rs#4614 class; not SMJ-specific). The KAN-25 sort-merge fallback routed over-budget join builds into this path and turned them into 600s wedges. - WEFT_SORT_MERGE_FALLBACK (default false) gates the SMJ fallback at all three sites; over-budget builds now run hash and fail fast with an actionable error naming the knob - Engine::smj_fallback_enabled(); regression tests for default-off hash planning and actionable pool-overflow errors; guard tests serialized behind a tokio Mutex; worker_join_memory_guard opts in - docs/runtime-contract.md + CFN + Helm values recommend preferHashJoin Live SF10: Q93 19.7s golden-clean (was >600s). Q11/Q72 remain — same upstream deadlock class via spilling sort/agg, needs the DataFusion fix. * fix(connect): capture + stamp lakehouse snapshot pins on distributed stages (KAN-48) The Spark Connect distributed path left StageDef.lakehouse_snapshot_pins empty, so workers (which arm require_lakehouse_snapshot_pins) rejected every lakehouse scan: delta/iceberg TPC-H SF10 was 0/22 while parquet was 22/22. try_run_distributed now plans via logical_plan_with_lakehouse_snapshots and stamps the pins JSON on every stage (mirroring the bench plan_distributed path); DataFrame relation-tree path wraps to_plan in the new Engine::capture_lakehouse_snapshots. Regression tests: crates/weft-connect/tests/distributed_lakehouse.rs — SQL + DataFrame distributed group-by over a delta table match single-node (pre-fix: exact production 'omitted the snapshot pin' error), and the unpinned-scan guard stays armed. * chore(deps): datafusion 54.0.0 -> 54.1.0 (KAN-47) Picks up apache/datafusion#23522 (backported as #23654): lost-wakeup race in SpillPoolWriter with concurrent writers — the exact mechanism behind our bounded-FairSpillPool stage deadlock at SF10 (TPC-DS Q11/Q72/Q93, the Q58/Q93 join-config tradeoff, and the KAN-45 SMJ wedges). Also #23574: SMJ correctness fix on the KAN-25 fallback path. Verified before commit: full gates green; upgrade spike ran the wedging Q93 SMJ plan at 4GiB bounded pool — wedged on 54.0.0, completed 4/4 on 54.1.0. datafusion-proto aligned to 54.1.0 for the RecursiveQuery.schema breaking change (#23886). * feat(execution,loom): stage no-progress watchdog + stage observability (KAN-47 follow-up) Defense-in-depth for the DF spill-pool deadlock class: a per-stage monitor races alongside the wall-clock timeout and driver-cancel flag inside run_stage_bounded, sampling three progress signals — per-task batch heartbeat (output stages now stream instead of collect so they heartbeat too), memory-pool activity timestamp (new ProgressMemoryPool wrapper), and DF+shuffle spill bytes. If none advance for WEFT_STAGE_NO_PROGRESS_SECS (default 600) the stage aborts with an actionable KAN-47 error and frees its task slot, instead of burning the full timeout silently. Observability: per-stage summary line (stage/partition, batches, spill bytes, duration, status, last-progress age) matching existing eprintln style; WEFT_STAGE_NO_PROGRESS_SECS + WEFT_STAGE_TIMEOUT_MS documented in docs/runtime-contract.md. Tests: stall-abort at ~2s (was 60s wall timeout), slow-but-progressing not aborted, signal-logic units, summary-line fields; summary lines verified in distributed test logs. * feat(loom,execution): per-query join strategy selection + stall-retry (KAN-53) WEFT_PREFER_HASH_JOIN is now tri-state: auto (new default) chooses per query — hash when the KAN-25 build-budget estimate fits, plan-time SMJ reroute when it doesn't (allowed in auto mode now that DF 54.1.0 fixed the SpillPoolWriter deadlock); true/false keep the legacy session-wide force. Stats fallback chain: row count x width -> parquet total_byte_size -> hash fast path backstopped by the runtime pool-exhaustion retry. Stall-retry: a stage aborted by the KAN-47 no-progress watchdog is cleared of partial spill and retried ONCE with the join strategy flipped (with_join_strategy_flipped task-local, flip accounts for the plan-time reroute so the retry really is the opposite strategy); driver cancels and wall-clock timeouts stay final. Also fixes a latent KAN-25 bug: the SMJ fallback's cloned SessionConfig registered a fresh empty default catalog, wiping engine tables after any fallback (with_create_default_catalog_and_schema(false)). Tests: env parsing/override, auto small-build hash / large-build SMJ, runtime retry on underreported estimate, flip inversion, distributed auto-mode guard variant, flight E2E stall-retry (completes after flip; always-stall fails after exactly two attempts). Gates green; tpch 22/22; tpcds 95/99 + 95/95 with Q58 AND Q93 both execute-verified. * fix(loom): auto join mode reroutes to SMJ when build estimate is unknown (KAN-53 follow-up) Live SF10 evidence: with unknown Glue/S3 stats, auto mode picked hash, the unaccounted hash build (KAN-57) ballooned past the pool, and workers were OOM-killed — TPC-H Q16/Q21 regressed from always-passing. The runtime pool-exhaustion retry can't backstop this because the build is not pool-accounted, so the OOM killer fires first. New fallback order (bounded pool, auto or KAN-45 opt-in): (1) over-budget estimate => plan-time SMJ reroute; (2) UNKNOWN estimate => SMJ reroute (safe since DF 54.1.0 fixed the SMJ deadlock); (3) positive under-budget estimate => hash; (4) runtime exhaustion => one SMJ retry. Unbounded pool keeps hash always (unchanged). Tests: UnknownStatsExec/UnknownStatsTable stand-ins; unknown+bounded => SMJ with correct results (fails before), unknown+unbounded => hash, positive-fit still hash; all prior KAN-25/53 tests unchanged. Gates green; tpch 22/22; tpcds 95/99 + 95/95 (Q58+Q93 verified). * fix(execution): compact StringView buffers in shuffle buckets; loud spilled-read errors Two latent shuffle-spill bugs exposed by TPC-H Q16 at SF10 on the new stack (auto join + DF 54.1.0): 1. hash_partition's arrow take produces StringViewArray views retaining the source scan's entire string buffers (~140x bloat: 33MB for 4,765 rows). estimated_bucket_bytes counted retained buffers, tripping the 4GB shuffle spill threshold; reading bloated buckets at 8-16 concurrent tasks climbed worker RSS to 31GB -> OOM-kill. Fix: compact_views() gc()s Utf8View columns after partitioning so buckets carry only live strings. 2. BucketCache::read_partition served spilled-bucket read errors as legitimately empty buckets (unwrap_or_default): a stage retry adjacent to the spill lifecycle silently truncated c0 values -> wrong supplier_cnt with exact row count. Fix: read_partition returns Result; missing base file is still a legitimate empty bucket, anything else fails the ShuffleRead pull loudly with stage/partition context. Tests: string_view_buckets_do_not_retain_source_buffers (fails on HEAD); spill read-back tests updated. Gates green; tpch 22/22; tpcds 95/99 + 95/95. Live SF10: Q16 3/3 consecutive golden-exact row diffs, zero spill, no OOM, stage tasks ~20ms (was 5-7s). * feat(execution): reorder filtered dims first in inner-join chains (Q72 spill blowup) TPC-DS Q72 at SF10 never finished (>1800s, 36-38GB sort spill per worker): with no table statistics DataFusion keeps written join order, so stage 0 exploded catalog_sales x inventory on item_sk alone into ~1.1B row pairs before the selective single-table filters could apply, then external-sorted that intermediate. New reorder_filtered_dims_first (planner rewrite in plan/join_order.rs, hooked at the top of plan_distributed_logical): within each contiguous inner-join chain under a Filter, leaves carrying single-table filter conjuncts join first (greedy, dependency-checked against ON columns; no-op when order is unchanged). Q72's chain shrinks to ~1e5 rows before inventory is touched: 48.96s, zero spill, golden row-diff clean (was >1800s). Tests: 5 unit tests (fires, idempotent, no-filter no-op, already-optimal no-op, non-inner bail). Gates green; tpch 22/22; tpcds 95/99 + 95/95. Live SF10: Q72 48.96s golden-clean; Q11/Q78/Q93 controls clean. * feat(site,scripts): SF10 distributed results on the site + cluster lifecycle tooling - site/src/data/tpch.json: TPC-H SF10 distributed (2 workers) — 22/22, total 533s, golden-validated vs DuckDB - site/src/data/tpcds.json: replaces pending placeholder — 60/99 execute (total 1314s, avg 21.9s), 39 clean strict-mode refusals, zero timeouts - bench/sf100/results/to-site-sf10.py: generator from suite jsonls to site JSONs (provenance: no hand-entered figures); ENGINE_COLORS entry - PerformancePage: SF10 distributed framing, honest TPC-DS 60/99 wording, reproduce commands - scripts/sf10-{start,stop,attach-spill}.sh: ASG scale-to-0 cost control and 200G spill volume attach (KAN-57) - golden check/diff harness used for all validations --------- Co-authored-by: Vamsi Krishna Kothapalli <vamsi@Vamsis-Mac-mini.local> Co-authored-by: Vamsi Krishna Kothapalli <vamsi@Vamsis-Mac-mini.attlocal.net> Co-authored-by: Vamsi Krishna Kothapalli <vamsi@vamsis-mac-mini.tail08ec2b.ts.net>

…xes (#63) * fix(bench,deploy): SF10-distributed harness, CFN ShufflePartitions, bootstrap fixes - register-glue.sh: fix SIGPIPE false-empty on multi-file table listings (grep -q closed the pipe early under pipefail; multi-file SF10 tables were falsely reported empty and skipped) - weft-cluster.yaml + deploy-stack.sh: ShufflePartitions parameter wired to the weft:shuffle-partitions tag (was hardcoded to WorkerCount) - remeasure-distributed.sh: SF-agnostic (SF env, GLUE_DB_* defaults derive from SF, --sf passthrough) - run-spark-connect.py: --query-timeout (env WEFT_QUERY_TIMEOUT) wraps collect in a ThreadPoolExecutor; on timeout the query fails fast and the Spark session is recreated so a wedged query can't hang the suite - packer bootstrap.sh: fix enable_role_unit deadlock + spill device (the shipped AMI still has the buggy copy; rebake pending) - docs: SF100/SF10 topology, memory invariant, WEFT_SHUFFLE_PARTITIONS - SF10-DISTRIBUTED-RESUME.md: live-cluster handoff state * fix(execution): distributed stage lifecycle, spill, and correctness fixes (KAN-15..33) - KAN-15/17: remove 30s Flight timeout, narrow is_retryable, WEFT_STAGE_TIMEOUT_MS, cancel path end-to-end - KAN-18/19: stage cleanup on all exit paths, ShuffleInputGuard, WEFT_STAGE_OUTPUT_TTL_SECS - KAN-20: completed_ops cap/TTL, real ReleaseExecute - KAN-21/23: worker-global spill budget + incremental .segN appends - KAN-22: correlated scalar subquery decorrelation (Q2 class) - KAN-25: hash-join build guard WEFT_HASH_JOIN_MAX_BUILD_FRACTION - KAN-26: Q12/Q13 join chains, agg-over-agg - KAN-27: one-row broadcast via __WEFT_SCALAR_STAGE__ literal injection - KAN-28: empty results return typed zero rows - KAN-29: semi/anti/IN decorrelation, Q17 residual, Q15 derived scalar, strict whole-fact-gather floor - KAN-32: intermediate bucket dispatch at 16 partitions/2 workers, streaming producers, per-partition dispatch, AQE row-count action - KAN-33: Q13/Q16/Q20/Q22 distributed shapes Regression tests: auto_distribute_decorrelate, auto_distribute_semi_anti, auto_distribute_kan33, distributed_high_card_groupby, stage_timeout, stage_cancel, stage_cleanup, worker_join_memory_guard. * fix(loom): key catalog provider cache by shard context (KAN-35) The catalog bridge cached resolved table providers by table name only, but a provider bakes in the shard/replicate decision from first resolution. Per-query auto-broadcast classification flips table roles between queries, so stale providers served full scans where the plan assumed sharded (rows x worker count — TPC-H Q4 exactly 2x at SF10) or shards where the plan assumed replicated (~0.5x loss in Q5/Q7, 0.25x in Q9). Cache is now keyed by (table_name, is_replicated). Regression test: crates/weft-loom/tests/catalog_shard_context.rs (role-flip in both directions; fails with the old name-only key). * fix(connect): distributed strict-mode surface + result-path alignment (KAN-15..33 support) * fix(bench): SF-agnostic remeasure harness, query-timeout flag, Q11 fraction (KAN-30) - run-spark-connect: --query-timeout, --skip-worker-preflight, jsonl results - q11.sql: HAVING fraction now 0.0001 / __WEFT_SF__ (TPC-H spec) - weft-bench: tpch-distributed WEFT_TPCH_ONLY/DEBUG, SF-agnostic remeasure - docs/runtime-contract.md updated to match * fix(execution): distribute Q13/Q22 under size-based auto-broadcast (KAN-36, KAN-38) The KAN-33 shapes only handled both-tables-sharded configs. The real cluster's resolve_replicated_tables replicates all but the largest table, so Q13/Q22 run with customer replicated and orders sharded: - Q13: relaxed broadcast for null-extended-side-sharded LEFT/RIGHT joins when every aggregate reads only null-extended-side columns (NULL extension contributes nothing to partials); routes the agg-over-agg composition at sharded.len()==1. 3-stage plan. - Q22: classify_scalar_conjunct accepts fully-replicated scalar bodies (partial runs once via ExchangeMode::Forward); body_projection allowed over fully-replicated outer, hash-shuffled by shared key. 6-stage plan. Regression tests: crates/weft-execution/tests/auto_distribute_kan36.rs (5 tests; fail before/pass after; 2-worker Flight e2e == single-node). tpch-distributed sf0.01: 22/22 distributed-ok, 0 mismatch. * fix(execution): serve only same-schema shuffle entries; lock AVG precision (KAN-39, KAN-40) KAN-39: Worker::read_shuffle unioned every stage-cache entry matching a stage id and took the first entry's schema. Stage ids repeat across queries, and a client-timed-out query's producer could insert after the driver's cleanup raced past it, so the next same-id stage served mixed-schema bucket sets and register_batches failed with 'Mismatch between schema and batches'. Now the freshest entry declares the served schema, benign nullability/metadata drift is aligned (align_batch_schema), and genuinely stale entries are skipped with a warning; plus a registration-side alignment guard in run_stage. KAN-40: not an engine bug — distributed AVG is bit-for-bit identical to single-node (Spark-faithful DECIMAL(19,6) typing); added distributed_avg.rs to lock distributed==single-node at full precision. Regression tests: stale_entry_from_prior_query_is_not_served (fails before the fix), crates/weft-execution/tests/distributed_avg.rs. * fix(execution): fuse grouped IN-subquery with outer aggregate (KAN-37) Q18's generic semi/anti plan shuffled the full 3-way customer-orders- lineitem join output (~60M wide rows at SF10) across 16 partitions, then grouped it all to discard everything but 624 orders — exceeding the 600s stage timeout. The outer sum(l_quantity) grouped by a key set containing o_orderkey IS the IN-subquery's per-key aggregate, so the fact never needs to join the dims: new try_in_agg_semi_join shape emits 3 stages (partial per-key agg -> combine+HAVING carrying r0 -> co-located join + exact local GROUP BY with sum(r0) tracking join multiplicity), strictly guarded so non-matching shapes fall through to the generic path. Live SF10: Q18 42.0s cold / 34.5s warm (was >600s), 100 rows, row-for-row exact vs DuckDB golden incl. ORDER BY/LIMIT boundary. Local sf3: 20.6s. Regression tests in auto_distribute_semi_anti.rs (plan shape fail-before, distributed==single-node, decline guards). * fix(execution): exact combine for agg-over-agg through CTE boundary (KAN-44) TPC-DS Q54 at SF10 split one customer's revenue across two shuffle partitions and never recombined by the group key: the single-sharded peels_to_inner_aggregate divert only fired for the KAN-36 null-extended outer-join shape, so Q54's inner GROUP BY c_customer_sk (inside the my_revenue CTE) fell through to the flat broadcast path and executed per worker over its local shard. The divert is now unconditional at sharded.len()==1 — partial hashed by inner key -> combine (one row per inner group) -> re-shuffle by outer key -> exact outer aggregate — subsuming the KAN-36 special case. Non-provable shapes (expression outer group keys, grouping sets) are declined instead of mis-planned. Regression tests: crates/weft-execution/tests/auto_distribute_kan44.rs (3 tests, fail before/pass after, distributed==single-node with customer rows spanning both shards). tpch-distributed 22/22 0 mismatch; TPC-DS execute ratchet 95/95 0 mismatch. * fix(connect): stream typed empty batch for zero-row distributed SQL results (KAN-42) The distributed SQL branch of base_relation_batches returned an empty Vec<RecordBatch> verbatim — every other result path honors the KAN-28 zero-row contract. For a zero-row distributed result (e.g. TPC-DS Q17), stream_batches emitted only ResultComplete: no ArrowBatch, no schema, no error, and PySpark died at 'assert table is not None'. Now mirrors the local path: re-derive the schema and push one typed empty batch. Regression test: distributed_zero_row_result_still_streams_typed_empty_batch (crates/weft-connect/tests/distributed_pyspark.rs) — fails before, passes after; asserts >=1 ArrowBatch response on a real ExecutePlan stream. * fix(execution): materialize replicated-only branches once; dedup identical CTE branches (KAN-41) TPC-DS Q58/Q78 inlined replicated-only aggregate branches (catalog_sales/ web_sales arms) in the per-partition outer stage, recomputing a full replicated-fact scan+join+aggregate 16x at SF10 (>600s); Q11 planned 4 structurally identical self-joined CTE branch DAGs (17 stages, each fact scanned 4x). dag_splitter now materializes replicated-only aggregate branches as single Forward stages (computed once; volatile branches and branches whose expression subqueries read sharded tables stay inline) and dedups identical CTE branches by plan fingerprint below SubqueryAlias. Q11: 17 -> 5 stages. Q72/Q93 investigated: no plan defect (already join-at-shard + partial aggregate; fast live-faithful) — harness artifact locally (all-MemTable workers lose stats/pushdown), slow-but-correct. Regression tests: crates/weft-execution/tests/auto_distribute_kan41.rs (5 tests, shape tests fail before/pass after) + 4 dag_splitter unit tests. Gates green; tpch-distributed 22/22; tpcds ratchets 95/99 + 95/95 held. * fix(loom,execution): make sort-merge fallback opt-in; fail fast over wedging (KAN-45) Root cause (local repro + live evidence): DataFusion 54 deadlocks under the bounded FairSpillPool — a spilling operator stalls while its consumer stops draining a RepartitionExec channel, parking the stage at 0% CPU until WEFT_STAGE_TIMEOUT_MS kills it (upstream delta-rs#4614 class; not SMJ-specific). The KAN-25 sort-merge fallback routed over-budget join builds into this path and turned them into 600s wedges. - WEFT_SORT_MERGE_FALLBACK (default false) gates the SMJ fallback at all three sites; over-budget builds now run hash and fail fast with an actionable error naming the knob - Engine::smj_fallback_enabled(); regression tests for default-off hash planning and actionable pool-overflow errors; guard tests serialized behind a tokio Mutex; worker_join_memory_guard opts in - docs/runtime-contract.md + CFN + Helm values recommend preferHashJoin Live SF10: Q93 19.7s golden-clean (was >600s). Q11/Q72 remain — same upstream deadlock class via spilling sort/agg, needs the DataFusion fix. * fix(connect): capture + stamp lakehouse snapshot pins on distributed stages (KAN-48) The Spark Connect distributed path left StageDef.lakehouse_snapshot_pins empty, so workers (which arm require_lakehouse_snapshot_pins) rejected every lakehouse scan: delta/iceberg TPC-H SF10 was 0/22 while parquet was 22/22. try_run_distributed now plans via logical_plan_with_lakehouse_snapshots and stamps the pins JSON on every stage (mirroring the bench plan_distributed path); DataFrame relation-tree path wraps to_plan in the new Engine::capture_lakehouse_snapshots. Regression tests: crates/weft-connect/tests/distributed_lakehouse.rs — SQL + DataFrame distributed group-by over a delta table match single-node (pre-fix: exact production 'omitted the snapshot pin' error), and the unpinned-scan guard stays armed. * chore(deps): datafusion 54.0.0 -> 54.1.0 (KAN-47) Picks up apache/datafusion#23522 (backported as #23654): lost-wakeup race in SpillPoolWriter with concurrent writers — the exact mechanism behind our bounded-FairSpillPool stage deadlock at SF10 (TPC-DS Q11/Q72/Q93, the Q58/Q93 join-config tradeoff, and the KAN-45 SMJ wedges). Also #23574: SMJ correctness fix on the KAN-25 fallback path. Verified before commit: full gates green; upgrade spike ran the wedging Q93 SMJ plan at 4GiB bounded pool — wedged on 54.0.0, completed 4/4 on 54.1.0. datafusion-proto aligned to 54.1.0 for the RecursiveQuery.schema breaking change (#23886). * feat(execution,loom): stage no-progress watchdog + stage observability (KAN-47 follow-up) Defense-in-depth for the DF spill-pool deadlock class: a per-stage monitor races alongside the wall-clock timeout and driver-cancel flag inside run_stage_bounded, sampling three progress signals — per-task batch heartbeat (output stages now stream instead of collect so they heartbeat too), memory-pool activity timestamp (new ProgressMemoryPool wrapper), and DF+shuffle spill bytes. If none advance for WEFT_STAGE_NO_PROGRESS_SECS (default 600) the stage aborts with an actionable KAN-47 error and frees its task slot, instead of burning the full timeout silently. Observability: per-stage summary line (stage/partition, batches, spill bytes, duration, status, last-progress age) matching existing eprintln style; WEFT_STAGE_NO_PROGRESS_SECS + WEFT_STAGE_TIMEOUT_MS documented in docs/runtime-contract.md. Tests: stall-abort at ~2s (was 60s wall timeout), slow-but-progressing not aborted, signal-logic units, summary-line fields; summary lines verified in distributed test logs. * feat(loom,execution): per-query join strategy selection + stall-retry (KAN-53) WEFT_PREFER_HASH_JOIN is now tri-state: auto (new default) chooses per query — hash when the KAN-25 build-budget estimate fits, plan-time SMJ reroute when it doesn't (allowed in auto mode now that DF 54.1.0 fixed the SpillPoolWriter deadlock); true/false keep the legacy session-wide force. Stats fallback chain: row count x width -> parquet total_byte_size -> hash fast path backstopped by the runtime pool-exhaustion retry. Stall-retry: a stage aborted by the KAN-47 no-progress watchdog is cleared of partial spill and retried ONCE with the join strategy flipped (with_join_strategy_flipped task-local, flip accounts for the plan-time reroute so the retry really is the opposite strategy); driver cancels and wall-clock timeouts stay final. Also fixes a latent KAN-25 bug: the SMJ fallback's cloned SessionConfig registered a fresh empty default catalog, wiping engine tables after any fallback (with_create_default_catalog_and_schema(false)). Tests: env parsing/override, auto small-build hash / large-build SMJ, runtime retry on underreported estimate, flip inversion, distributed auto-mode guard variant, flight E2E stall-retry (completes after flip; always-stall fails after exactly two attempts). Gates green; tpch 22/22; tpcds 95/99 + 95/95 with Q58 AND Q93 both execute-verified. * fix(loom): auto join mode reroutes to SMJ when build estimate is unknown (KAN-53 follow-up) Live SF10 evidence: with unknown Glue/S3 stats, auto mode picked hash, the unaccounted hash build (KAN-57) ballooned past the pool, and workers were OOM-killed — TPC-H Q16/Q21 regressed from always-passing. The runtime pool-exhaustion retry can't backstop this because the build is not pool-accounted, so the OOM killer fires first. New fallback order (bounded pool, auto or KAN-45 opt-in): (1) over-budget estimate => plan-time SMJ reroute; (2) UNKNOWN estimate => SMJ reroute (safe since DF 54.1.0 fixed the SMJ deadlock); (3) positive under-budget estimate => hash; (4) runtime exhaustion => one SMJ retry. Unbounded pool keeps hash always (unchanged). Tests: UnknownStatsExec/UnknownStatsTable stand-ins; unknown+bounded => SMJ with correct results (fails before), unknown+unbounded => hash, positive-fit still hash; all prior KAN-25/53 tests unchanged. Gates green; tpch 22/22; tpcds 95/99 + 95/95 (Q58+Q93 verified). * fix(execution): compact StringView buffers in shuffle buckets; loud spilled-read errors Two latent shuffle-spill bugs exposed by TPC-H Q16 at SF10 on the new stack (auto join + DF 54.1.0): 1. hash_partition's arrow take produces StringViewArray views retaining the source scan's entire string buffers (~140x bloat: 33MB for 4,765 rows). estimated_bucket_bytes counted retained buffers, tripping the 4GB shuffle spill threshold; reading bloated buckets at 8-16 concurrent tasks climbed worker RSS to 31GB -> OOM-kill. Fix: compact_views() gc()s Utf8View columns after partitioning so buckets carry only live strings. 2. BucketCache::read_partition served spilled-bucket read errors as legitimately empty buckets (unwrap_or_default): a stage retry adjacent to the spill lifecycle silently truncated c0 values -> wrong supplier_cnt with exact row count. Fix: read_partition returns Result; missing base file is still a legitimate empty bucket, anything else fails the ShuffleRead pull loudly with stage/partition context. Tests: string_view_buckets_do_not_retain_source_buffers (fails on HEAD); spill read-back tests updated. Gates green; tpch 22/22; tpcds 95/99 + 95/95. Live SF10: Q16 3/3 consecutive golden-exact row diffs, zero spill, no OOM, stage tasks ~20ms (was 5-7s). * feat(execution): reorder filtered dims first in inner-join chains (Q72 spill blowup) TPC-DS Q72 at SF10 never finished (>1800s, 36-38GB sort spill per worker): with no table statistics DataFusion keeps written join order, so stage 0 exploded catalog_sales x inventory on item_sk alone into ~1.1B row pairs before the selective single-table filters could apply, then external-sorted that intermediate. New reorder_filtered_dims_first (planner rewrite in plan/join_order.rs, hooked at the top of plan_distributed_logical): within each contiguous inner-join chain under a Filter, leaves carrying single-table filter conjuncts join first (greedy, dependency-checked against ON columns; no-op when order is unchanged). Q72's chain shrinks to ~1e5 rows before inventory is touched: 48.96s, zero spill, golden row-diff clean (was >1800s). Tests: 5 unit tests (fires, idempotent, no-filter no-op, already-optimal no-op, non-inner bail). Gates green; tpch 22/22; tpcds 95/99 + 95/95. Live SF10: Q72 48.96s golden-clean; Q11/Q78/Q93 controls clean. * feat(site,scripts): SF10 distributed results on the site + cluster lifecycle tooling - site/src/data/tpch.json: TPC-H SF10 distributed (2 workers) — 22/22, total 533s, golden-validated vs DuckDB - site/src/data/tpcds.json: replaces pending placeholder — 60/99 execute (total 1314s, avg 21.9s), 39 clean strict-mode refusals, zero timeouts - bench/sf100/results/to-site-sf10.py: generator from suite jsonls to site JSONs (provenance: no hand-entered figures); ENGINE_COLORS entry - PerformancePage: SF10 distributed framing, honest TPC-DS 60/99 wording, reproduce commands - scripts/sf10-{start,stop,attach-spill}.sh: ASG scale-to-0 cost control and 200G spill volume attach (KAN-57) - golden check/diff harness used for all validations * feat(execution): distribute subquery-over-sharded shapes (KAN-55) - resolve_replicated_tables now sizes subquery-scanned tables too, so per-query classification is uniform (subquery-only facts replicate) - replicated-subquery conjuncts route verbatim (partition-independent) - global aggregates above semi/anti filters: recombinable partials + partition-0 gate for COUNT(DISTINCT) (exact empty-input row) - Q9 scalar-projection shape: uncorrelated global-agg scalars over the sharded fact become per-worker partials + one-row combine, AST-rewritten projection evaluated once behind a partition-0 gate - anti-only inline guard: NOT EXISTS over sharded key stream with replicated outer now declines to gather (was per-partition duplicates) 6 of 8 queries now plan distributed at SF10 strict (Q9,Q10,Q16,Q35,Q69, Q94); Q95 (self-join IN) and Q14 (ROLLUP+INTERSECT) keep refusing with locking tests. 15 new tests in auto_distribute_kan55.rs, all fail-before. Gates green; tpch 22/22; tpcds ratchets 95/99 + 95/95. * feat(execution): distribute mixed sharded/replicated UNION ALL (KAN-54) Nested-union flattening in split_union_by_sharding and plan_union (bag union is associative), plus a SUM-only guard for pre-aggregated sharded arms. 6 queries newly plan distributed under the real SF10 strict classification (Q4,Q27,Q33,Q56,Q60,Q76), all execute-verified distributed==single-node. Q5/Q77/Q80 deliberately keep refusing (grouping-set composition has a documented wrong-answer trap). Tests: auto_distribute_kan54.rs (7 tests, fail-before). Remainder mapped: window-over-agg ROLLUP family next (Q36,Q44,Q47,Q51,Q57,Q67,Q70,Q86). * fix(execution,connect): cancel+cleanup on client disconnect; reap uncommitted spill (KAN-46, KAN-31) Dropping the driver query future mid-run (client disconnect) skipped all cleanup: cancel watcher leaked, worker stages kept running, buckets stayed cached (17.8-26.5GB idle RSS observed), and uncommitted producer spill (25-38GB) orphaned into later ENOSPC. QueryAbortGuard on Drop aborts the watcher and runs the cancel+clear sweep; StageSpillReaper reaps a stage's spill scope (base + .segN segments) whenever a producer exits uncommitted. Committed caches still await clear_stages/TTL (positive control). Tests: query_abort_cleanup.rs (4), client_disconnect.rs (2) incl. slow-but-connected client is NOT killed. All fail-before. * test(loom): pin Spark default null ordering (KAN-52) Spark default: ASC => NULLS FIRST, DESC => NULLS LAST (opposite of DuckDB/Postgres). weft already matched via datafusion.sql_parser.default_null_ordering=nulls_min; this dedups the config block and locks the contract with 7 tests (defaults, explicit overrides, ORDER BY...LIMIT boundary, window ORDER BY, distributed finalize replay). Verified RED when the config is disabled. * feat(bench): MATCH/BENIGN/MISMATCH golden verdicts; parquet ListingTable harness (KAN-50, KAN-51) golden_common.py: verdict ladder — checksum match, verified numeric-scale (round-half-even/trunc at decimal scale, reproduces the recorded hash), boundary-tie (LIMIT + tie-zone enumeration, null-ordering sensitivity), numeric-drift heuristic, MISMATCH; --self-test (27 checks) and --strict-benign for CI. TPC-H final: 20 MATCH + 2 BENIGN + 0 MISMATCH; TPC-DS final: 47 MATCH + 13 BENIGN + 0 MISMATCH. tpcds_dist workers register replicated tables as parquet ListingTables (real stats, dynamic-filter pushdown — live-faithful) instead of MemTables; Q72 harness run 51.6s -> 17.8s at SF10. Ratchets 95/99 + 95/95 held; tpch 22/22. * fix(deploy): close bootstrap systemd cycle; atomic env; symmetric DNS; name-agnostic spill (KAN-58) Bootstrap no longer issues role-unit start jobs from inside weft-bootstrap.service (the cycle-closing edge; Requires=/After= kept as the env+DNS-before-start guarantee; UserData starts the role unit after bootstrap completes). write_env is atomic (mktemp+mv). Worker DNS: boot sync always includes own IP (fixes missing-self on cold start); ExecStop removes only own IP (fixes stale zombie IPs -> 'no free task slots'). find_spill_device accepts any unmounted non-root disk, largest wins. TimeoutStartSec=600 for the 240s peer wait. Offline harness 23/23; bash -n clean. NOTE: AMI rebake needed to roll this out. * fix(execution): nested-IN semi cascade accepts replicated dims (KAN-55 follow-up) KAN-55's uniform classification (subquery tables sized too) flips Q20's supplier/partsupp/part from sharded-by-default to replicated at SF10, and try_nested_in_semi's shardedness checks declined — strict mode refused with a whole-lineitem gather. Relaxed: replicated outer/middle/nested scans export once via ExchangeMode::Forward (KAN-36 precedent); the correlated scalar body still requires a sharded fact (a replicated body would multiply the threshold by worker count). Tests: q20_sf10_replicated_dims_plans_semi_cascade (fail-before at the SF10 config: 8 stages, Forward on 2/3/6, no gather) + q20_sf10_replicated_dims_distributed_matches_single_node. Gates green; tpch 22/22; tpcds 95/99 + 95/95; all KAN-55 suites re-verified. * feat(site): TPC-DS 72/99 + TPC-H re-run on the KAN-49 wave-2 stack Site data regenerated from the wave-2 validation runs: TPC-DS 72/99 execute (was 60/99; 27 strict refusals remain), TPC-H 22/22 re-run. to-site-sf10.py now reads the wave-2 jsonls (Q20 dedup ok-preferred). --------- Co-authored-by: Vamsi Krishna Kothapalli <vamsi@Vamsis-Mac-mini.local> Co-authored-by: Vamsi Krishna Kothapalli <vamsi@Vamsis-Mac-mini.attlocal.net> Co-authored-by: Vamsi Krishna Kothapalli <vamsi@vamsis-mac-mini.tail08ec2b.ts.net>

github-actions Bot added the physical-plan Changes to the physical-plan crate label Jul 13, 2026

pepijnve changed the title ~~Improve robustness of SpillPoolReader with multiple concurrent writers~~ Resolve lost wakeup in SpillPoolReader with multiple concurrent SpillPoolWriters Jul 13, 2026

pepijnve mentioned this pull request Jul 13, 2026

Release DataFusion 54.1.0 (minor/patch) Release #22547

Closed

31 tasks

jayzhan211 reviewed Jul 14, 2026

View reviewed changes

pepijnve force-pushed the issue_23447 branch from 518c07f to 1042dea Compare July 14, 2026 14:04

Improve robustness of SpillPoolReader with multiple concurrent writers

191cceb

pepijnve force-pushed the issue_23447 branch from 1042dea to 191cceb Compare July 14, 2026 14:09

alamb reviewed Jul 15, 2026

View reviewed changes

alamb approved these changes Jul 15, 2026

View reviewed changes

Update documentation

c50bc33

adriangb and others added 2 commits July 15, 2026 12:43

Add MPSC usage test

7cb38fb

adriangb mentioned this pull request Jul 15, 2026

Enforce single-writer spill pools for preserve_order via the type system pepijnve/datafusion#3

Merged

Merge pull request #3 from pydantic/spill-pool-typed-writers

73db692

Enforce single-writer spill pools for preserve_order via the type system

pepijnve added 5 commits July 15, 2026 20:32

Fix test compile error

363b280

Merge remote-tracking branch 'upstream/main' into issue_23447

b8f9719

# Conflicts: # datafusion/physical-plan/src/repartition/mod.rs

Invert SharedSpillPoolWriter/SpillPoolWriter relationship to eliminat…

2f2c52d

…e SpillWriter wrapper type

Formatting

ac6dcb3

Rename shared_writer to new_writer

1e5d680

pepijnve added 2 commits July 15, 2026 21:36

More documentation tweaks

37075ad

Fix typo in docstring

bb67abd

pepijnve added 5 commits July 16, 2026 09:40

Eliminate panic code path

79d4648

Fix compile error

64e6125

Revert name changes to avoid backwards incompatible change

f5032f1

Update docstrings

8972148

Documentation updates

fc84765

jayzhan211 reviewed Jul 16, 2026

View reviewed changes

jayzhan211 approved these changes Jul 16, 2026

View reviewed changes

alamb approved these changes Jul 16, 2026

View reviewed changes

Merge remote-tracking branch 'apache/main' into issue_23447

536055c

github-actions Bot added the auto detected api change Auto detected API change label Jul 16, 2026

fmt

1d46d57

fix links

c1c3d88

pepijnve mentioned this pull request Jul 16, 2026

[branch-54] Resolve lost wakeup in SpillPoolReader with multiple conc… #23654

Merged

alamb added this pull request to the merge queue Jul 16, 2026

Merged via the queue into apache:main with commit 12fa0ce Jul 16, 2026
40 checks passed

hareshkh reviewed Jul 17, 2026

View reviewed changes

	// Don't put back current_write_file - let it rotate
	// Don't put back current_write_files - let it rotate

Uh oh!

Conversation

pepijnve commented Jul 13, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

SemyonSinchenko commented Jul 13, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pepijnve Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

pepijnve commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

adriangb commented Jul 15, 2026

Uh oh!

pepijnve commented Jul 15, 2026

Uh oh!

pepijnve commented Jul 15, 2026

Uh oh!

adriangb commented Jul 15, 2026

Uh oh!

pepijnve commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pepijnve commented Jul 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pepijnve commented Jul 15, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alamb commented Jul 16, 2026

Uh oh!

alamb left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

pepijnve commented Jul 16, 2026

Uh oh!

codecov-commenter commented Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

alamb commented Jul 16, 2026

Uh oh!

pepijnve commented Jul 16, 2026

Uh oh!

alamb commented Jul 16, 2026

Uh oh!

pepijnve commented Jul 16, 2026

pepijnve Jul 14, 2026 •

edited

Loading

pepijnve commented Jul 15, 2026 •

edited

Loading

pepijnve commented Jul 15, 2026 •

edited

Loading

pepijnve commented Jul 15, 2026 •

edited

Loading

github-actions Bot commented Jul 16, 2026 •

edited

Loading

codecov-commenter commented Jul 16, 2026 •

edited

Loading