[PDGGK]

Description

A fragment instance (FI) of a query that has already succeeded can stay in FLUSHING when its consumer closes the local channel first. It keeps its admission memory until SimpleQueryTerminator finishes it about one second later, so a query sent in that second can be rejected with 719: There is not enough memory to execute current fragment instance.

I found this through the intermittent 719s of IoTDBLTTBTableFunctionIT in #18725. With this one-line change on that PR's head, the class passed 8 runs in a row with -PTableSimpleIT, against 9 failing runs out of 12 without it. Its -PTableClusterIT failures have a second cause, which this change does not address.

To check that it does not depend on that PR, I reproduced it on master 999721fcaf with a local IT (not part of this PR), without LTTB:

• one DataNode (operator pool 16,090,037 B), 4 data region groups, sort_buffer_size_in_bytes = 3980000;
• two queries alternate 300 times, each on a fresh connection that reads its whole result:
  • SELECT part, count(s1) AS c, max(s1) AS m FROM regions GROUP BY part ORDER BY c DESC, part, whose root FI reserves 4,001,832 B;
  • SELECT time, part, s1 FROM regions ORDER BY part, time, whose four region FIs and root need 15,950,312 B, so it is rejected only if something else still holds more than 139,725 B.

	runs	runs with a 719
master	5	5
this change	5	0 (1,500 of 1,500 queries succeed)

At the first 719 of every master run, the memory held by other queries was the 4,001,832 B root FI of the preceding, already completed query, plus at most 9,352 B. (After that first rejection, master kept rejecting queries until the run ended, 89 to 295 of 300 per run; I have not looked into that part.)

Cause

1. The consumer closes first: when a finished query's result handle is closed, or when a LIMIT stops reading (ExchangeOperator.java:144). LocalSourceHandle.close() → SharedTsBlockQueue.close() → LocalSinkChannel.close() (SharedTsBlockQueue.java:370).
2. LocalSinkChannel.close() sets closed = true and then calls its listener (LocalSinkChannel.java:230, :233).
3. The listener calls ShuffleSinkHandle.close() (MPPDataExchangeManager.java:629, :635), which calls channel.close() again. That call returns false because the channel is already closed (LocalSinkChannel.java:228), so the handle skips sinkListener.onFinish(this) (ShuffleSinkHandle.java:280, :291-:292).
4. context.finished() (MPPDataExchangeManager.java:529) is not called, and the FI stays in FLUSHING until SimpleQueryTerminator, after its 1000 ms grace period, calls FragmentInstanceManager.cancelTask, which calls context.finished() for a query that did not fail (SimpleQueryTerminator.java:96-:97, :115; FragmentInstanceManager.java:411-:414).

The remote SinkChannel.close() calls invokeOnFinished() before it sets closed = true (SinkChannel.java:366-:367), so on that path the handle does finish.

Change

selfClosed = channel.close() || (channel.isClosed() && !channel.isAborted());

A channel that is already closed and not aborted now counts as closed; an aborted channel still does not. The FI then reaches FINISHED through the same context.finished() call that the terminator makes a second later. When the consumer stops early, as with a LIMIT, the FI now finishes at that point, as it already does on the remote path: the producer's later send() and setNoMoreTsBlocks() return immediately, and its driver ends at the next sink.isClosed() check (Driver.java:224).

With several channels, selfClosed still takes the result of the last channel, as before; I left that unchanged. A mix of aborted and consumer-closed channels does not reach this close() through the channel listener, which skips it once any channel has aborted (MPPDataExchangeManager.java:631).

Verification

• New tests in MPPDataExchangeManagerTest, with the queue, channels, source handle and ShuffleSinkHandle created through MPPDataExchangeManager and only FragmentInstanceContext mocked:
  • the consumer closing right after no-more is set (forced with a direct listener on isBlocked(), no sleeps);
  • the consumer closing before no-more;
  • two channels, both closed by their consumers;
  • the remote consumer closing first;
  • an aborted channel, then close() on the handle;
  • an aborted channel next to a consumer-closed one, where the handle must stay open.
• The three local consumer-close tests fail on master with ShuffleSinkHandle did not finish after the consumer closed ...; the other three pass on master too and guard the existing behaviour. All nine tests in the class pass with the change. Without !channel.isAborted(), testAbortedLocalChannelKeepsShuffleSinkHandleFromFinishing fails.
• All test classes under queryengine/execution/exchange and queryengine/execution/fragment pass (20 classes, 84 tests), and so do spotless:check and the -P with-zh-locale compile.
• ITs with the change, all passing:
  • -PTableSimpleIT (140 tests, 33 of them @Ignored in IoTDBOrderByTableIT): IoTDBCollectFanInIT, IoTDBTableAggregationIT, IoTDBOrderByTableIT, IoTDBOrderByLimitOffsetAlignByDeviceTableIT, IoTDBAlignedOffsetLimitPushDownTableIT;
  • -PSimpleIT (96 tests): IoTDBShuffleSink1IT, IoTDBShuffleSink2IT, IoTDBOrderByIT, IoTDBOrderByLimitOffsetAlignByDeviceIT, IoTDBAggregationIT;
  • -PTableClusterIT (70 tests): IoTDBCollectFanInIT, IoTDBTableAggregationIT, IoTDBOrderByLimitOffsetAlignByDeviceTableIT.

---

This PR has:

• been self-reviewed.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage.