[experiment] Runtime hash join build-side selection on StageBoundary - #12

Which issue does this PR close?

Not meant for merge. This is an experiment on top of apache#25798 to test the idea in apache#25798 (comment): use StageBoundary to pick a hash join's build side at runtime. Related to apache#23194.

Rationale for this change

The planner chooses a hash join's build side from static statistics, which are often wrong. This PR checks whether the StageBoundary API is enough to fix that choice at runtime on a real benchmark.

What changes are included in this PR?

Hack-quality code, organized as two commits:

1. demo
   • Adds ExecutionPlan::as_boundary(), defaulting to None, so a driver can find boundaries by walking the plan.
   • Adds a default method StageBoundary::progress(partition) -> Option<StageProgress>, which reports rows, bytes and whether the input finished. It also changes is_ready to mean "safe to inspect".
   • Adds a toy example: cargo run --release --example execution_monitoring -- adaptive_join.
2. hack: runtime hash join build-side selection driver
   • ThresholdBoundaryExec buffers each partition up to a byte limit that the driver can raise. Buffered memory is charged to the memory pool.
   • AdaptiveJoinExec sits at the plan root. Before executing the plan, it resolves joins from the bottom up. For each join, it races both inputs, doubling the limit, until one input finishes and the other is known to have at least as many rows. If the input that finished is the probe side, it swaps the join.
   • The AdaptiveJoinBuildSide optimizer rule is behind datafusion.optimizer.adaptive_join_build_side, which is off by default.

What is the testing strategy for this PR?

There are two unit tests for the swap safety check. Beyond those, it was tested by running TPC-H at SF100 on a 32-core box with dfbench, flag off and on.

• Correctness: Q4, Q8, Q9, Q10, Q15 and Q22 return identical results with the flag off and on.

• Memory: the clearest result is on Q9, where the planner puts 150M-row and 80M-row inputs on the build side against a 32M-row probe.

  Q9	Static plan	Adaptive plan
  Smallest --memory-limit that completes	14G	9G
  Peak RSS, no limit	17.0 GB	12.2 GB
  Time	~12.7 s	~12.6 s

• Speed: there is no meaningful speedup. The full suite ranges from neutral to 3.6% slower, because each join now waits for both of its inputs to be drained before it starts. The toy example is 6x faster, but only because its small side fits in cache and it runs on a single partition. TPC-H's misestimated joins all have two large inputs.

• Whole suite at a 10G limit: 20 of 22 queries pass with the flag off and with it on. The flag rescues Q9 but makes Q7 fail.

Things this turned up:

• HashJoinExec keeps reporting hash partitioning on its join keys after its embedded projection drops them.
• get_array_memory_size counts StringView data buffers once per batch that shares them, so after a RepartitionExec it overcounts by up to 32x.
• Scalar subqueries need their own driver, because the subquery values have to exist before the main input is primed.

Are there any user-facing changes?

Two new experimental config options, both off by default. The config docs and information_schema test have not been updated.

🤖 Generated with Claude Code