[BugFix] fix MulticastSinkOperator stuck in OUTPUT_FULL state (backport #67153)

[mergify]

Why I'm doing:

example sql:

set cbo_cte_reuse = true;
set cbo_cte_reuse_rate = 0;

with cte as (
	select * from table limit 2000000
) 
select k1, min(c1) from cte group by k1
union all
select k1, min(c2) from cte group by k1

    "queries_in_workgroup": [
        {
            "query_id": "d668e65c-dfec-11f0-bd33-00163e1bdf0b",
            "fragments": [
                {
                    "fragment_id": "d668e65c-dfec-11f0-bd33-00163e1bdf13",
                    "fragment_status": "OK",
                    "drivers": [
                        {
                            "driver_id": 1,
                            "state": "OUTPUT_FULL",
                            "driver_desc": "query_id=d668e65c-dfec-11f0-bd33-00163e1bdf0b fragment_id=d668e65c-dfec-11f0-bd33-00163e1bdf13 driver=driver_1_1 addr=0x1528cec29d10, status=OUTPUT_FULL, operator-chain: [exchange_source_1_0x1528cec27f10(O) { has_output:true} -> limit_1_0x15291f2b3110(O) -> multi_cast_local_exchange_sink_1_0x15290e0d8f90(O)]"
                        }
                    ]
                }
            ]
        }
    ]

In a query plan such as multi-cast -> limit, the completion of the source operator should also trigger the completion of the sink operator; otherwise, it may remain in the OUTPUT_FULL state indefinitely.

One practical query is ANALYZE SAMPLE <table>, as it produces a CTE query similar to the one mentioned above.

What I'm doing:

Fixes #issue

What type of PR is this:

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Does this PR entail a change in behavior?

• Yes, this PR will result in a change in behavior.
• No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

• Interface/UI changes: syntax, type conversion, expression evaluation, display information
• Parameter changes: default values, similar parameters but with different default values
• Policy changes: use new policy to replace old one, functionality automatically enabled
• Feature removed
• Miscellaneous: upgrade & downgrade compatibility, etc.

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr needs user documentation (for new or modified features or behaviors)
  • I have added documentation for my new feature or new function
• This is a backport pr

Bugfix cherry-pick branch check:

• I have checked the version labels which the pr will be auto-backported to the target branch
  • 4.0
  • 3.5
  • 3.4
  • 3.3

---

Note

Prevents multicast sink from hanging when upstream sources (e.g., after a limit) complete by short‑circuiting sink completion.

• Adds is_short_circuit() to MultiCastLocalExchanger and uses it in sink is_finished()
• Implements short‑circuit in InMemoryMultiCastLocalExchanger (_opened_source_number == 0)
• No behavior changes elsewhere; header-only updates to exchanger and sink operator

^{Written by Cursor Bugbot for commit 0923f66. This will update automatically on new commits. Configure here.}

---

This is an automatic backport of pull request #67153 done by [Mergify](https://mergify.com).

[mergify]

Cherry-pick of 8d4ed82 has failed:

On branch mergify/bp/branch-3.4/pr-67153
Your branch is up to date with 'origin/branch-3.4'.

You are currently cherry-picking commit 8d4ed82e01.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	modified:   be/src/exec/pipeline/exchange/mem_limited_chunk_queue.cpp
	modified:   be/src/exec/pipeline/exchange/mem_limited_chunk_queue.h
	modified:   be/src/exec/pipeline/exchange/multi_cast_local_exchange.cpp
	modified:   be/src/exec/pipeline/exchange/multi_cast_local_exchange.h
	modified:   be/src/exec/pipeline/exchange/multi_cast_local_exchange_sink_operator.h
	modified:   be/src/exec/pipeline/exchange/spillable_multi_cast_local_exchange.cpp
	new file:   test/sql/test_spill/R/test_spill_mcast_hang
	new file:   test/sql/test_spill/T/test_spill_mcast_hang

Unmerged paths:
  (use "git add/rm <file>..." as appropriate to mark resolution)
	deleted by us:   be/src/exec/pipeline/exchange/split_local_exchange.cpp
	deleted by us:   be/src/exec/pipeline/exchange/split_local_exchange.h

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

[mergify]

@mergify[bot]: Backport conflict, please reslove the conflict and resubmit the pr

[mergify]

🧪 CI Insights

Here's what we observed from your CI run for 4cd34fc.

🟢 All jobs passed!

But CI Insights is watching 👀