You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by fe...@apache.org on 2022/12/03 14:23:19 UTC
[incubator-kyuubi] branch branch-1.6 updated: [KYUUBI #3885][1.6] Fix memory leak when using incremental collect mode
This is an automated email from the ASF dual-hosted git repository.
feiwang pushed a commit to branch branch-1.6
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
The following commit(s) were added to refs/heads/branch-1.6 by this push:
new 2b015b93b [KYUUBI #3885][1.6] Fix memory leak when using incremental collect mode
2b015b93b is described below
commit 2b015b93b19bfbd8b0fe10c026be8255e7642cbe
Author: fwang12 <fw...@ebay.com>
AuthorDate: Sat Dec 3 22:23:10 2022 +0800
[KYUUBI #3885][1.6] Fix memory leak when using incremental collect mode
Fix memory leak when using incremental mode.
FYI: https://stackoverflow.com/questions/45649044/scala-stream-iterate-and-memory-management
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible
- [X] Add screenshots for manual tests if appropriate before:
 after:
- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request
Closes #3885 from wForget/dev.
Closes #3885
4f1b1bffb [wForget] add test
129ac1efa [wForget] Fix memory leak when using incremental mode
Authored-by: wForget <643348094qq.com>
Signed-off-by: fwang12 <fwang12ebay.com>
Closes #3895 from turboFei/mem_leak.
Closes #3885
765652a76 [fwang12] [KYUUBI #3885] Fix memory leak when using incremental collect mode
Authored-by: fwang12 <fw...@ebay.com>
Signed-off-by: fwang12 <fw...@ebay.com>
---
.../engine/spark/operation/ExecuteStatement.scala | 4 +-
.../spark/operation/SparkOperationSuite.scala | 48 ++++++++++++++++++++++
2 files changed, 51 insertions(+), 1 deletion(-)
diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala
index 0b0e438f3..313c6330b 100644
--- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala
+++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala
@@ -95,7 +95,9 @@ class ExecuteStatement(
iter =
if (incrementalCollect) {
info("Execute in incremental collect mode")
- new IterableFetchIterator[Row](result.toLocalIterator().asScala.toIterable)
+ new IterableFetchIterator[Row](new Iterable[Row] {
+ override def iterator: Iterator[Row] = result.toLocalIterator().asScala
+ })
} else {
val resultMaxRows = spark.conf.getOption(OPERATION_RESULT_MAX_ROWS.key).map(_.toInt)
.getOrElse(session.sessionManager.getConf.get(OPERATION_RESULT_MAX_ROWS))
diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala
index 2ac0235c6..81e63ab3b 100644
--- a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala
+++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/operation/SparkOperationSuite.scala
@@ -31,6 +31,7 @@ import org.apache.spark.kyuubi.SparkContextHelper
import org.apache.spark.sql.catalyst.analysis.FunctionRegistry
import org.apache.spark.sql.types._
+import org.apache.kyuubi.config.KyuubiConf
import org.apache.kyuubi.engine.SemanticVersion
import org.apache.kyuubi.engine.spark.WithSparkSQLEngine
import org.apache.kyuubi.engine.spark.schema.SchemaHelper.TIMESTAMP_NTZ
@@ -333,6 +334,53 @@ class SparkOperationSuite extends WithSparkSQLEngine with HiveMetadataTests with
}
}
+ test("test fetch orientation with incremental collect mode") {
+ val sql = "SELECT id FROM range(2)"
+
+ withSessionConf(Map(KyuubiConf.OPERATION_INCREMENTAL_COLLECT.key -> "true"))()() {
+ withSessionHandle { (client, handle) =>
+ val req = new TExecuteStatementReq()
+ req.setSessionHandle(handle)
+ req.setStatement(sql)
+ val tExecuteStatementResp = client.ExecuteStatement(req)
+ val opHandle = tExecuteStatementResp.getOperationHandle
+ waitForOperationToComplete(client, opHandle)
+
+ // fetch next from before first row
+ val tFetchResultsReq1 = new TFetchResultsReq(opHandle, TFetchOrientation.FETCH_NEXT, 1)
+ val tFetchResultsResp1 = client.FetchResults(tFetchResultsReq1)
+ assert(tFetchResultsResp1.getStatus.getStatusCode === TStatusCode.SUCCESS_STATUS)
+ val idSeq1 = tFetchResultsResp1.getResults.getColumns.get(0)
+ .getI64Val.getValues.asScala.toSeq
+ assertResult(Seq(0L))(idSeq1)
+
+ // fetch next from first row
+ val tFetchResultsReq2 = new TFetchResultsReq(opHandle, TFetchOrientation.FETCH_NEXT, 1)
+ val tFetchResultsResp2 = client.FetchResults(tFetchResultsReq2)
+ assert(tFetchResultsResp2.getStatus.getStatusCode === TStatusCode.SUCCESS_STATUS)
+ val idSeq2 = tFetchResultsResp2.getResults.getColumns.get(0)
+ .getI64Val.getValues.asScala.toSeq
+ assertResult(Seq(1L))(idSeq2)
+
+ // fetch prior from second row, expected got first row
+ val tFetchResultsReq3 = new TFetchResultsReq(opHandle, TFetchOrientation.FETCH_PRIOR, 1)
+ val tFetchResultsResp3 = client.FetchResults(tFetchResultsReq3)
+ assert(tFetchResultsResp3.getStatus.getStatusCode === TStatusCode.SUCCESS_STATUS)
+ val idSeq3 = tFetchResultsResp3.getResults.getColumns.get(0)
+ .getI64Val.getValues.asScala.toSeq
+ assertResult(Seq(0L))(idSeq3)
+
+ // fetch first
+ val tFetchResultsReq4 = new TFetchResultsReq(opHandle, TFetchOrientation.FETCH_FIRST, 3)
+ val tFetchResultsResp4 = client.FetchResults(tFetchResultsReq4)
+ assert(tFetchResultsResp4.getStatus.getStatusCode === TStatusCode.SUCCESS_STATUS)
+ val idSeq4 = tFetchResultsResp4.getResults.getColumns.get(0)
+ .getI64Val.getValues.asScala.toSeq
+ assertResult(Seq(0L, 1L))(idSeq4)
+ }
+ }
+ }
+
test("get operation status") {
val sql = "select date_sub(date'2011-11-11', '1')"