Finish query only after the results are fully consumed

[arhimondr]

This commit makes it consistent with QueryState#FINISHED documentation.

This change is needed to allow spooling exchange to be used for query
results delivery. Spooling exchanges in fault tolerant execution are
closed on query completion. It is important not to close an exchange
before the results are fully consumed.

Is this change a fix, improvement, new feature, refactoring, or other?

Improvement

Is this a change to the core query engine, a connector, client library, or the SPI interfaces? (be specific)

Core engine

How would you describe this change to a non-technical end user or system administrator?

With this change a query is considered finished successfully only when the query results are successfully consumed by the client. Before it was possible that a query might've been marked as successful even when client experienced a failure or dropped a connection.

Related issues, pull requests, and links

Documentation

(x) No documentation is needed.
( ) Sufficient documentation is included in this PR.
( ) Documentation PR is available with #prnumber.
( ) Documentation issue #issuenumber is filed, and can be handled later.

Release notes

( ) No release notes entries required.
(x) Release notes entries required with the following suggested text:

# Core
* Queries are now marked as finished successfully only when the query results are successfully consumed by the client

[arhimondr]

Why this change is needed?

In fault tolerant execution we've been using a streaming (direct) exchange to deliver query results to coordinator. With #13025 query results will be delivered using a spooling exchange. Spooling exchanges are closed (wiped out) upon query completion (either success or a failure). It is important not to close a spooling exchange before the query results are fully consumed. To achieve that it is required to only transition a query to a FINISHED state when the results are fully consumed.

This change also makes the behavior consistent with the QueryState#FINISHED documentation:

trino/core/trino-main/src/main/java/io/trino/execution/QueryState.java

Line 52 in 6cb3199

* Query has finished executing and all output has been consumed.

[losipiuk]

What would happen if a client on purpose reads only a fraction of the result. I think this may be a common use case that a UI tool interacting with Trino (over JDBC or ODBC) just reads a fixed number of rows, up to some limit, even if the result is larger. Would query be marked as FAILED then in Trino? It should not be.

[arhimondr]

@losipiuk it will be marked as CANCELED. The only difference is that with this change when results are not read completely the query will be marked as CANCELED always, and before this change it would be marked as FINISHED if all the stages finished running but there are still query results in the buffer.

[losipiuk]

looks like this should be separate commit.

[losipiuk]

Or I guess I do not fully understand this change. Can you explain?

[arhimondr]

That's a tricky one. For queries with no output stage (e.g.: DDL operations such as DROP TABLE) the exchange is still created. It has to be closed to allow query to finish, otherwise the client will be waiting of ExchangeClient#isBlocked and the query will never finish. The client is closed by the closeExchangeClientIfNecessary. However with this change the query will not transition to the FINISHED state before the results are consumed. So we have to close the ExchangeClient when a query is in FINISHING state to allow the client to pool the results and mark the query as resultsConsumed to allow it to finish.

[losipiuk]

LGTM but let's sync so I am sure I understand it correctly.

[losipiuk]

Actually I prefered previous shape with explanation you put in the PR; maybe just convert it to comment in the code.
No we are closing exchange on ANY state change - which is weird. The question which arises why do we need to open it at all in the first place.

[arhimondr]

No we are closing exchange on ANY state change - which is weird.

Yeah, it is weird. For DDL operations it can only be closed once query info is available (to figure out there is no output stage). Essentially this is simply to wait for the query info to become available.

The question which arises why do we need to open it at all in the first place.

I think it is mostly for convenience (creating ExchangeClient lazily is likely to add more complexity)

[arhimondr]

Added a comment

[arhimondr]

CI: #1671

[losipiuk]

please squash before merging.