[Bug] Partitioned hash join under enable_force_spill neither publishes nor disables runtime filters, consumer scan blocks for full runtime_filter_wait_time_ms

[alexey-bass]

Search before asking

• I had searched in the issues and found no similar issues.

Version

Reproduced identically on two versions (VeloDB Cloud builds of Doris):

• selectdb-4.1.3-39dbf23ff3d (Doris 3.1.4)
• cloud-4.1.8-737218b90e2 (Doris 3.1.7) - upgraded specifically to re-test, same behavior

Deployment: compute-storage decoupled (cloud mode), single BE, parallel_fragment_exec_instance_num=2.

What's Wrong?

A trivial LEFT JOIN with LIMIT 100 hangs in RUNNING state for the entire
runtime_filter_wait_time_ms (900 s in our setup) before returning, when the
following session variables are set:

enable_force_spill=true
enable_join_spill=true
enable_agg_spill=true
enable_sort_spill=true

Query:

SELECT t.*, srstart.stop_name AS actual_start_stop
FROM trip_time_convert t
LEFT JOIN stop_on_route_stop_master srstart
  ON srstart.id = t.actual_start_stop_id
LIMIT 100;

Table sizes: probe/preserved side ~2.9M rows, other side ~9.79M rows.
The planner rewrites this to RIGHT_OUTER_JOIN, pushes LIMIT 100 down to the
build side, and attaches two runtime filters to the 9.79M-row scan:
RF000[min_max] and RF001[in_or_bloom] on the join key.

From the profile:

1. Because enable_force_spill=true, the join runs as
   RIGHT OUTER JOIN(PARTITIONED) even though the build side is only 100 rows.
2. The build completes in ~2 ms. Two build instances: InputRows = 100 and 0
   (one instance receives zero rows). No spill actually occurs - every Spill*
   counter is 0.
3. The runtime filters are never produced:
   PublishRuntimeFilterTime = 0ns, RuntimeFilterComputeTime = 0ns.
4. They are also never disabled: profile shows Disabled = false,
   Ignored = false.
5. As a result the consumer scan never starts: NumScanners = 0, ScanRows = 0,
   both filters stuck in RuntimeFilterState = NOT_READY, synced_size = -1,
   has_remote_target = true, WaitTimeMS = 900000. The query "completes" only
   when the RF wait times out.

Key profile excerpt (identical on both versions):

OLAP_SCAN_OPERATOR (id=2, table = stop_on_route_stop_master):
  RuntimeFilter: (id = 0, type = minmax):
    Info: [Id = 0, IsPushDown = false, RuntimeFilterState = NOT_READY,
           HasRemoteTarget = true, HasLocalTarget = false,
           Ignored = false, Disabled = false, WaitTimeMS = 900000]
  RuntimeFilter: (id = 1, type = in_or_bloomfilter):
    Info: [Id = 1, ... RuntimeFilterState = NOT_READY, ... Disabled = false]
  NumScanners: 0
  ScanRows: 0

PARTITIONED_HASH_JOIN_SINK_OPERATOR (id=4):
  InputRows: 100 (instance 0) / 0 (instance 1)
  ExecTime: ~2ms, all Spill* counters = 0

PARTITIONED_HASH_JOIN_PROBE_OPERATOR (id=4), BuildPhase:
  RuntimeFilterComputeTime: 0ns
  PublishRuntimeFilterTime: 0ns

Full profiles from both versions attached.

What You Expected?

Per the documented spill design, the partitioned hash join should either
(a) publish the runtime filters normally when no spill occurs, or
(b) on spilling, disable the RF and signal the consumer so the scan proceeds
without waiting.

The observed state is a third, invalid one: RF neither published nor disabled,
leaving the consumer blocked for the full wait time. Even though
enable_force_spill is a debug/testing flag, it should not deadlock a query
that never actually spills.

Expected result for this query: sub-second. Verified: with
enable_force_spill=false (and the other spill variables off), the identical
query returns in 90 ms and the join runs as a plain HASH_JOIN with both RFs
reaching READY.

How to Reproduce?

1. Two tables joined on an id column, big enough that the planner attaches
   min_max + in_or_bloom runtime filters to the larger side's scan
   (~10M rows on the RF target side in our case).
2. Set parallel_fragment_exec_instance_num=2 (so that with LIMIT pushdown one
   build instance can end up with 0 input rows - suspected trigger for the RF
   merge never completing, since both RFs have remote targets and
   synced_size stays -1).
3. SET enable_force_spill=true; SET enable_join_spill=true;
4. Run: SELECT t.*, d.name FROM fact t LEFT JOIN dim d ON d.id = t.dim_id LIMIT 100;
5. Query hangs until runtime_filter_wait_time_ms elapses.
6. SET enable_force_spill=false; rerun - returns immediately.

Repro rate: 100% with the variables above, on both tested versions.

Anything Else?

• Profile IDs for reference:
  • 3.1.4 run: c6cd07df59084b3d-8da5afe29dfce262
  • 3.1.7 run: 8dca88a596834a4b-aacef345cb7ab612
• Workaround confirmed: disable enable_force_spill (900 s -> 90 ms).

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[alexey-bass]

3.1.4 - Query ID c6cd07df59084b3d-8da5afe29dfce262.txt
3.1.7 - Query ID 8dca88a596834a4b-aacef345cb7ab612.txt
3.1.7 (with workaround) - Query ID ccb9f123e824420a-83366882ada8c34d.txt

[Doris-Breakwater]

Initial triage: confirmed BE runtime-filter lifecycle bug

This is a valid bug with high confidence from source inspection at both exact build commits reported in the issue (39dbf23ff3d and 737218b90e2). The observed 900-second pause is a timeout-bounded liveness/latency failure; there is no evidence here of an incorrect query result.

Verified code path

1. RuntimeState::enable_join_spill() returns true when either enable_force_spill or enable_join_spill is true.
2. For a non-broadcast hash join, PipelineFragmentContext::_create_operator() then selects PartitionedHashJoin* and explicitly executes tnode_.runtime_filters.clear() under the comment TODO: support rf in partitioned hash join.
3. In these affected revisions, that RF-cleared tnode_ is also used to initialize the build-side inner HashJoinBuildSinkOperatorX. Consequently, the BE registers no RF producer for this join, while the RF consumer descriptors already attached to the scan node are unaffected.
4. With no registered producer, there is nothing to compute, publish, or disable. This directly explains RuntimeFilterComputeTime = 0ns, PublishRuntimeFilterTime = 0ns, consumer state NOT_READY, Disabled = false, and waiting until runtime_filter_wait_time_ms expires.
5. This happens when the partitioned operator is constructed, before any runtime spill decision. Therefore, the all-zero spill counters are expected and do not contradict the diagnosis.

The zero-row second build instance is not the primary trigger: RF descriptors are removed for every instance. The code predicts that the same orphaned consumer can occur with parallel_fragment_exec_instance_num=1, provided the join remains non-broadcast/partitioned and the RF target is remote. LIMIT 100 mainly makes it easy to demonstrate the non-spilling branch.

The scope is also wider than the debug flag alone: enable_join_spill=true selects the same path even when enable_force_spill=false.

Recommended fix

Please backport the complete RF lifecycle used by newer partitioned-hash-join code, rather than only removing runtime_filters.clear():

1. Preserve the original join tnode (including RF descriptors) for the build-side inner HashJoinBuildSinkOperatorX; use the RF-cleared copy only for per-partition recovery/probe-side inner operators.
2. On the no-spill path, wire the wrapper's finish dependency to the inner build sink and have PartitionedHashJoinSinkLocalState::close() close the inner sink, so its normal RF size/build/publish path runs.
3. On the first transition to the spilled path, call HashJoinBuildSinkLocalState::disable_runtime_filters() before any empty-build early return. That method sends size zero, marks the filters disabled, and publishes the terminal state so remote consumers are released.

Only doing step 1 is insufficient: in the affected code, the wrapper does not close the inner sink on the no-spill path, and _revoke_unpartitioned_block() does not disable RFs on the spill path. Conversely, merely clearing BE producer descriptors is unsafe unless FE also removes every corresponding consumer descriptor.

Regression coverage

Please cover all of the following:

• Partitioned join selected, no actual spill, one build instance: RF reaches READY and the scan does not wait for the RF timeout.
• Two build instances with one receiving zero rows: every producer participates in size synchronization/publication.
• Actual build spill: RF reaches a terminal disabled/ignored state and the scan starts immediately.
• Both selection modes independently: enable_join_spill=true with force spill off, and enable_force_spill=true.
• A remote RF target, since a local-only target may not expose this failure.

Additional information useful for the regression test

The current issue body says the full profiles are attached, but its raw Markdown contains no attachment links. Please reattach the exported profiles if the upload was lost. A minimal DDL/data generator plus EXPLAIN VERBOSE would also make the exact RIGHT OUTER JOIN/LIMIT/remote-RF topology deterministic in regression testing. These items are not blockers for the root-cause determination above.

Current workaround: keep both enable_force_spill and enable_join_spill disabled for affected builds (or reduce runtime_filter_wait_time_ms only as a mitigation; that does not resolve the orphaned RF lifecycle).

The issue currently has no labels; it should be triaged under BE execution, spill, and runtime filter.

Breakwater-GitHub-Analysis-Slot: slot_49f75bc8b5f0