[FEATURE] Incremental result fetching for Flink engine

[link3280]

Code of Conduct

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

Search before asking

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

Describe the feature

Add incremental result fetching for Flink engine.

Motivation

Currently, Flink query results are buffered in 3 layers:

1. Flink result fetcher
2. Flink engine
3. Kyuubi server

That leads to unnecessary memory footprints and adds more latency. Moreover, the data consistency could be affected since every layer is of different size limits.

Describe the solution

No response

Additional context

No response

Are you willing to submit PR?

• Yes. I would be willing to submit a PR with guidance from the Kyuubi community to improve.
• No. I cannot submit a PR at this time.

[link3280]

Currently, the beeline incremental fetching works on the client side. The Kyuubi server pulls all result rows from the engine and returns them to the client in a micro-batch manner.

However, this could be problematic for Flink engine or other streaming scenarios, where the operation is still running and producing more records after the Kyuubi operation is executed asynchronizedly and considered result-available.

Consider a select-from-kafka use case, if we simply forward to fetch result requests to the Flink engine, there would be 3 situations:

• the Kafka topic is temporarily empty, we get an empty rowset
• the Kafka topic has 10 records, so we get a rowset with 10 rows (less than kyuubi.session.engine.flink.max.rows)
• the Kafka topic has lots of records, so we get a rowset with ${kyuubi.session.engine.flink.max.rows} rows

In the first two situations, Flink engine should but can't tell the Kyuubi server to fetch again later, thus the current workaround is that Flink engine polls result set until the rows reached ${kyuubi.session.engine.flink.max.rows}.

We could fix this by adding a hasMoreResults field to the result fetching response (i.e. TRowSet), but this would cause a large footprint, as it touches the thrift protocol. So I'm thinking of injecting TRowSet with metadata columns prefixed with __KYUUBI_ to allow the engines to tell the server there're more rows to fetch. The server needs to convert these metadata columns into TFetchResultsResp fields and drop them before returning the TRowSet to the clients.

kyuubi/kyuubi-common/src/main/scala/org/apache/kyuubi/service/TFrontendService.scala

Lines 530 to 533 in 056c5ef

	val rowSet = be.fetchResults(operationHandle, orientation, maxRows, fetchLog)
	resp.setResults(rowSet)
	resp.setHasMoreRows(false)
	resp.setStatus(OK_STATUS)

WDYT? @pan3793 @bowenliang123

[pan3793]

The Spark engine already supported engine-side incremental collection, I think there is no blocker on the infrastructure

[link3280]

The difference is that in Spark scenarios the result is always ready and complete, so the client would know the result rows are fully fetched if one fetch is empty. That's to say, TFetchResultsResp.hasMoreRows is not required (and it's not used actually, see the codes below).

kyuubi/kyuubi-hive-jdbc/src/main/java/org/apache/kyuubi/jdbc/hive/KyuubiQueryResultSet.java

Lines 333 to 348 in 056c5ef

	if (fetchedRows == null || !fetchedRowsItr.hasNext()) {
	TFetchResultsReq fetchReq = new TFetchResultsReq(stmtHandle, orientation, fetchSize);
	TFetchResultsResp fetchResp;
	fetchResp = client.FetchResults(fetchReq);
	Utils.verifySuccessWithInfo(fetchResp.getStatus());
	TRowSet results = fetchResp.getResults();
	fetchedRows = RowSetFactory.create(results, protocol);
	fetchedRowsItr = fetchedRows.iterator();
	}
	if (fetchedRowsItr.hasNext()) {
	row = fetchedRowsItr.next();
	} else {
	return false;
	}

However, in Flink scenarios, the fetched result could be temporarily empty because the result data is not ready. That means we need a clear sign of EOS, which could be TFetchResultsResp.hasMoreRows.

[bowenliang123]

Any progress in this feature is welcomed. But hasMoreRows may be heavy for it. For the first two situations, the server could just wait for the timeout (with a new config added for this) and then return the resultset with the statement closed, without footprints in the whole fetching process.

[link3280]

@bowenliang123 Thanks for your input! IMHO, introducing a timeout seems to be a workaround for this problem but doesn't solve the problem at the root cause.

For flink jobs with heavy states or high parallelism, it might take minutes for job initialization, it's likely that we fetch no rows before the timeout. Users could adjust the timeout of course, but it's hard to determine an appropriate timeout threshold for a single job without a few tries.

I think hasMoreRows is not as heavy as we thought (BTW, hasMoreRows is an existing field that's not put into use), as it's false by default and should be compatible with batch/olap engines. WDYT?

[bowenliang123]

SGTM. In my priority, 1. timed-out config for testing / adhoc one-time query situation (fetching stable results within user-confident timeout), 2. hasMoreRows for incremental result fetching. They can be implemented to satisfy either or both scenarios, while the first one could be easier. If the second one is a better solution for you, I'm good with it.

[link3280]

@bowenliang123 Now I'm in favor of the simple timeout solution you proposed. The more I coded the more I realized the divergence between the current result fetching and the streaming incremental one.

The former assumes the results are ready, so when result hasNext()==true, users expect that there's always a row returned by next(). However, there's no guarantee of a new row for streaming incremental fetch, so hasNext()==true means there COULD BE a new row within a fetch timeout. That makes confusion for the semantics of result fetching and may surprise users.

I'm closing this issue in flavor of #5088.

[bowenliang123]

Yes, thanks for considering my suggestion. hasNext does hint at an infinite result set with an unstable size, in certain scenarios with a common unbound stream. Let's see whether the timed-out approach suite for most cases.

[link3280]

Reopen the issue because after a discussion with @pan3793 , we think the semantic problem is acceptable:

• For Beeline users, beeline supports incremental mode which pulls rows in a micro-batch manner. If no row is produced, Beeline will block on result.next() until there are some rows or the timeout is reached (no timeout by default).
• For JDBC users, calling result.next() now could be long-blocking, and users could either set a fetch timeout or interrupt the thread externally.

However, we still need to think about what to do if the timeout happens. There should be 3 approaches:

1. throw a timeout exception with result.next()
2. return false for result.next() which indicates there are no more rows (which is technically not true).
3. return true for result.next() but the current row is set to null which indicates there are no rows yet (which is against normal use cases and may lead to surprises).

Personally, I lean toward approach 1.

[pan3793]

@link3280 thanks for the summary of the offline discussion, yes, I agree with you, throwing a dedicated exception is a better approach.

[bowenliang123]

+1 for approach 1.