[SPARK-50235][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec #48767

viirya · 2024-11-05T22:10:34Z

What changes were proposed in this pull request?

This patch cleans up ColumnVector resource after processing all rows in ColumnarToRowExec. This patch only focus on codeben implementation of ColumnarToRowExec. For non-codegen, it should be relatively rare to use, and currently no good way has proposed, so leaving it to a follow up.

Why are the changes needed?

Currently we only assign null to ColumnarBatch object but it doesn't release the resources hold by the vectors in the batch. For OnHeapColumnVector, the Java arrays may be automatically collected by JVM, but for OffHeapColumnVector, the allocated off-heap memory will be leaked.

For custom ColumnVector implementations like Arrow-based, it also possibly causes issues on memory safety if the underlying buffers are reused across batches. Because when ColumnarToRowExec begins to fill values for next batch, the arrays in previous batch are still hold.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing tests.

Was this patch authored or co-authored using generative AI tooling?

No

…ToRowExec

viirya · 2024-11-05T22:17:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

+       |if ($batch != null) {
+       |  $batch.close();
+       |}


This looks like memory leak for off-heap case.

viirya · 2024-11-05T22:17:59Z

sql/core/src/main/scala/org/apache/spark/sql/execution/Columnar.scala

@@ -194,9 +194,14 @@ case class ColumnarToRowExec(child: SparkPlan) extends ColumnarToRowTransition w
       |    $shouldStop
       |  }
       |  $idx = $numRows;
+       |  $batch.closeIfNotWritable();


For writable column vectors, they are reused across batches, so we cannot close them until finishing all batches (see below).

…l rows in ColumnarToRowExec ### What changes were proposed in this pull request? This patch cleans up ColumnVector resource after processing all rows in ColumnarToRowExec. This patch only focus on codeben implementation of ColumnarToRowExec. For non-codegen, it should be relatively rare to use, and currently no good way has proposed, so leaving it to a follow up. ### Why are the changes needed? Currently we only assign null to ColumnarBatch object but it doesn't release the resources hold by the vectors in the batch. For OnHeapColumnVector, the Java arrays may be automatically collected by JVM, but for OffHeapColumnVector, the allocated off-heap memory will be leaked. For custom ColumnVector implementations like Arrow-based, it also possibly causes issues on memory safety if the underlying buffers are reused across batches. Because when ColumnarToRowExec begins to fill values for next batch, the arrays in previous batch are still hold. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No Closes #48767 from viirya/close_if_not_writable. Authored-by: Liang-Chi Hsieh <viirya@gmail.com> Signed-off-by: Kent Yao <yao@apache.org> (cherry picked from commit 800faf0) Signed-off-by: Kent Yao <yao@apache.org>

yaooqinn · 2024-11-06T09:12:59Z

Merged to master and 3.5

Thank you @viirya @dongjoon-hyun

viirya · 2024-11-06T14:47:21Z

Thanks @dongjoon-hyun @yaooqinn

bersprockets · 2024-12-05T03:28:18Z

Starting with this commit (800faf0), I get an error with the following commands:

val testDf = spark.range(200000).selectExpr("id as a", "concat('x', string(id % 2)) as b")
testDf.write.mode("overwrite").partitionBy("b").format("parquet").save("test1")

spark.read.parquet("test1").createOrReplaceTempView("test1")
sql("select * from test1 limit 12 offset 20000").collect

The collect results in this error:

Exception in task 0.0 in stage 2.0 (TID 17)
java.lang.NullPointerException: Cannot invoke "org.apache.spark.unsafe.types.UTF8String.getBaseObject()" because "input" is null
	at org.apache.spark.sql.catalyst.expressions.codegen.UnsafeWriter.write(UnsafeWriter.java:111) ~[spark-catalyst_2.13-4.0.0-SNAPSHOT.jar:4.0.0-SNAPSHOT]
	at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source) ~[?:?]
	at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) ~[spark-sql_2.13-4.0.0-SNAPSHOT.jar:4.0.0-SNAPSHOT]
	at org.apache.spark.sql.execution.WholeStageCodegenEvaluatorFactory$WholeStageCodegenPartitionEvaluator$$anon$1.hasNext(WholeStageCodegenEvaluatorFactory.scala:50) ~[spark-sql_2.13-4.0.0-SNAPSHOT.jar:4.0.0-SNAPSHOT]

If I build with the commit previous to 800faf0, I get actual results rather than an error.

viirya · 2024-12-05T03:54:15Z

It was fixed by #49021.

dongjoon-hyun · 2024-12-10T16:15:54Z

Just for the record, Iceberg community seems to report a bug against this patch.

Bump Spark 3.5.4 RC1 iceberg#11731

viirya · 2024-12-10T17:24:15Z

Thanks @dongjoon-hyun. Please see #49131 (comment) for some discussion.

dongjoon-hyun · 2024-12-10T18:25:08Z

Thank you for clarifying that swiftly!

Close not writable ColumnVector after processing all rows in Columnar…

c1abdcd

…ToRowExec

github-actions bot added the SQL label Nov 5, 2024

viirya changed the title ~~[SPARK-XXXXX][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec~~ [SPARK-50235][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec Nov 5, 2024

viirya commented Nov 5, 2024

View reviewed changes

viirya mentioned this pull request Nov 5, 2024

Spark ColumnarToRowExec cannot pass CometBuffer safety check apache/datafusion-comet#1059

Open

dongjoon-hyun approved these changes Nov 6, 2024

View reviewed changes

yaooqinn approved these changes Nov 6, 2024

View reviewed changes

yaooqinn closed this in 800faf0 Nov 6, 2024

viirya deleted the close_if_not_writable branch November 6, 2024 14:47

LuciferYang mentioned this pull request Dec 10, 2024

[SPARK-50463][SQL] Fix ConstantColumnVector with Columnar to Row conversion #49021

Closed

pan3793 mentioned this pull request Dec 10, 2024

Bump Spark 3.5.4 RC1 apache/iceberg#11731

Draft

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-50235][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec #48767

[SPARK-50235][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec #48767

viirya commented Nov 5, 2024 •

edited

Loading

viirya Nov 5, 2024

viirya Nov 5, 2024

yaooqinn commented Nov 6, 2024

viirya commented Nov 6, 2024

bersprockets commented Dec 5, 2024

viirya commented Dec 5, 2024

dongjoon-hyun commented Dec 10, 2024 •

edited

Loading

viirya commented Dec 10, 2024

dongjoon-hyun commented Dec 10, 2024

[SPARK-50235][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec #48767

[SPARK-50235][SQL] Clean up ColumnVector resource after processing all rows in ColumnarToRowExec #48767

Conversation

viirya commented Nov 5, 2024 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Was this patch authored or co-authored using generative AI tooling?

viirya Nov 5, 2024

Choose a reason for hiding this comment

viirya Nov 5, 2024

Choose a reason for hiding this comment

yaooqinn commented Nov 6, 2024

viirya commented Nov 6, 2024

bersprockets commented Dec 5, 2024

viirya commented Dec 5, 2024

dongjoon-hyun commented Dec 10, 2024 • edited Loading

viirya commented Dec 10, 2024

dongjoon-hyun commented Dec 10, 2024

viirya commented Nov 5, 2024 •

edited

Loading

dongjoon-hyun commented Dec 10, 2024 •

edited

Loading