[Query] Share table fragment exchange budgets and consume ready Collect inputs

[ColinLeeo]

Description

Per-handle exchange quotas and sequential Collect consumption can leave ready inputs waiting while other inputs hold unused capacity. This change enables ready-input Collect/MappingCollect consumption together with shared exchange quotas only within eligible table-model fragment instances (FIs). Tree-model queries and excluded table FIs retain sequential Collect consumption and independent quotas.

Exchange accounting

• Decide eligibility after local operator generation, before estimating root operator memory. Require table Collect/MappingCollect, allow only Collect/Union as multi-input nodes, and stop traversal at Exchange boundaries. Keep independent plan-node quotas and sequential Collect consumption for FIs with MergeSort, Join, unsupported multi-input nodes, any TopK (including its single-child form), internal pipelines, or scheduling dependencies. The root pipeline factory is registered after memory estimation; any previously registered child pipeline excludes sharing.
• FI context owns inputMaxBytes and outputMaxBytes. Give the full F to the only active direction; split INPUT = F / 2, OUTPUT = F - INPUT when both directions charge the FI. Unused directions get zero. Remote SourceHandles charge INPUT; fragment sinks and local queues charge their producing FI's OUTPUT. Fall back if an active direction cannot hold the configured maximum TsBlock.
• Initialize all shared handles from the FI's corresponding limit. Preserve the public legacy per-handle quota calculation and the original tree-model initialization path. Restore DEBUG logs for independent quotas and add localized FI INPUT/OUTPUT shared-limit logs. MemoryPool keeps shared/independent accounting mode and usage counters, and checks the limit supplied with each request. Pending requests retain that limit for retries. The pool retains no FI context or FI budget fields; physical global capacity still bounds allocations.
• Check quotas and update accounting under the FI lock with bounded atomic global allocation. Complete waiting futures outside that lock and retain the retry after publishing a waiter, including cancellation synchronization.

Collect and plan properties

• Keep Collect sequential by default. Register table Collect/Union operators during generation and opt them into ready-input consumption only when their complete FI passes the budget and dependency checks. Choose the mode before memory estimation so retained-state accounting matches execution. Preserve the ready-queue optimization: activate live direct children once, retain asynchronous completions in a queue, and reobserve only the advanced child instead of scanning all inputs per block. Reuse pending-future listeners, preserve a selection during output backpressure, and call operators only on the driver thread. Refresh readiness after null/empty results and propagate failures and cancellation.
• Retain MappingCollect's child-specific mapping and account for all active children. Clear invalid pre-grouping above unordered Collect/Union, preserve valid streaming aggregation in ordered branches, restore required sorting, and prevent sort elimination across unordered fan-in.
• Keep OperatorTreeGenerator and visitTopK() identical to the PR base. TopK planning and runtime filter ID handling, along with MergeSort and Join consumption algorithms, remain unchanged. TopK-specific fixes are deferred; this PR excludes those FIs from the new mode.

SQL assertions

• Preserve the existing TestUtils ResultSet assertion overloads. Add explicit CollectQueryAssertions for the affected tests, using the existing SQL parser to inspect final-result ordering before accepting unordered assertions.
• Reject final ORDER BY and arbitrary top-level LIMIT/OFFSET on the unordered path. Nested UNION branches, subqueries, window ordering, comments, and string literals are distinguished through the AST. Compare every node independently and preserve duplicate multiplicity.
• Add explicit outer ORDER BY to tests that require deterministic order. EXPLAIN checks use a separate guarded path to validate each node's plan semantics.

Validation

• 147 targeted unit tests passed: 19 Collect tests and 128 DataNode tests, covering FI initialization, consistent handle limits, supplied request limits, physical pool bounds, sequential defaults, table-only opt-in, single-child TopK and pipeline fallback, retained-memory accounting, concurrent admission, wakeups, cancellation, and cleanup.
• Seven deterministic SQL assertion regressions passed, covering final versus nested ordering, arbitrary slices, duplicate counts, and every-node checks.
• The full English reactor build (including test compilation and targeted unit tests) and Chinese test-compile -DskipTests passed.
• 23 targeted integration cases passed in both TableSimpleIT and TableClusterIT (three DataNodes), using the same rebuilt node template. Compatibility checks cover original tree-model device/time ordering and unchanged TopK results above a Collect path.
• Spotless, Checkstyle, license checks, and git diff --check passed. Maven validation used offline mode.

---

This PR has:

• been self-reviewed, including concurrent reservation, release, cancellation, and readiness paths.
• documented modified behaviors in the PR description and code comments.
• added comments explaining concurrency and ownership decisions.
• added or updated unit tests.
• added integration tests.
• been tested in a test IoTDB cluster.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Copilot]

🔵 Needs a closer look

An unresolved block-size boundary affects exchange startup, and the change spans concurrent accounting and distributed query planning.

1 open finding

• Fixed memory split can prevent exchange block reservation · New

🧠 Review effort: Lite

[Copilot]

🟡 Changes recommended

Shared-budget admission and dependency handling can leave queries stalled despite otherwise available progress paths.

4 open findings

• Shared OUTPUT budget can deadlock dependent pipeline forwarding · New
• Non-atomic reservations can leave concurrent requests permanently pending · New
• MergeSort can deadlock under shared INPUT memory pressure · New
• Fixed memory split can prevent exchange block reservation

🧠 Review effort: Balanced

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.