Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[SPARK-38273][SQL] decodeUnsafeRows's iterators should close underlying input streams #35613

Closed
wants to merge 11 commits into from
Closed

[SPARK-38273][SQL] decodeUnsafeRows's iterators should close underlying input streams #35613

Changes from 5 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
ef600eb
[SPARK-38273][SQL] Fix native memory leak in SparkPlan
kevins-29 Feb 22, 2022
5b240ac
retrigger checks
kevins-29 Feb 22, 2022
64da846
retrigger checks
kevins-29 Feb 22, 2022
db9f4a4
cleaning up how we add NextIterator as TaskCompletionListener
kevins-29 Feb 22, 2022
4afc8b5
setting finished=true on exception
kevins-29 Feb 22, 2022
fb67692
removing addTaskCompletionListener. catching Thowable instream of Exc…
kevins-29 Feb 23, 2022
5ca1e78
minor refactor to make backporting easier
kevins-29 Feb 24, 2022
8a8db10
minor refactor to make backporting easier
kevins-29 Feb 24, 2022
68d95b8
Update sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPl…
kevins-29 Feb 24, 2022
b23be3c
retrigger checks
kevins-29 Feb 24, 2022
5662cfa
git commit on suggestion left trailing white space
kevins-29 Feb 24, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 26 additions & 10 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/SparkPlan.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger

import scala.collection.mutable.{ArrayBuffer, ListBuffer}

import org.apache.spark.{broadcast, SparkEnv}
import org.apache.spark.{broadcast, SparkEnv, TaskContext}
import org.apache.spark.internal.Logging
import org.apache.spark.io.CompressionCodec
import org.apache.spark.rdd.{RDD, RDDOperationScope}
Expand All @@ -37,6 +37,7 @@ import org.apache.spark.sql.errors.QueryExecutionErrors
import org.apache.spark.sql.execution.metric.SQLMetric
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.vectorized.ColumnarBatch
import org.apache.spark.util.NextIterator

object SparkPlan {
/** The original [[LogicalPlan]] from which this [[SparkPlan]] is converted. */
Expand Down Expand Up @@ -384,17 +385,32 @@ abstract class SparkPlan extends QueryPlan[SparkPlan] with Logging with Serializ
val bis = new ByteArrayInputStream(bytes)
val ins = new DataInputStream(codec.compressedInputStream(bis))

new Iterator[InternalRow] {
new NextIterator[InternalRow] {
Option(TaskContext.get()).foreach(_.addTaskCompletionListener[Unit](_ => closeIfNeeded()))
kevins-29 marked this conversation as resolved.
Show resolved Hide resolved
private var sizeOfNextRow = ins.readInt()
kevins-29 marked this conversation as resolved.
Show resolved Hide resolved
override def hasNext: Boolean = sizeOfNextRow >= 0
override def next(): InternalRow = {
val bs = new Array[Byte](sizeOfNextRow)
ins.readFully(bs)
HyukjinKwon marked this conversation as resolved.
Show resolved Hide resolved
val row = new UnsafeRow(nFields)
row.pointTo(bs, sizeOfNextRow)
sizeOfNextRow = ins.readInt()
row
override def getNext(): InternalRow = {
kevins-29 marked this conversation as resolved.
Show resolved Hide resolved
if (sizeOfNextRow >= 0) {
kevins-29 marked this conversation as resolved.
Show resolved Hide resolved
try {
val bs = new Array[Byte](sizeOfNextRow)
ins.readFully(bs)
val row = new UnsafeRow(nFields)
row.pointTo(bs, sizeOfNextRow)
sizeOfNextRow = ins.readInt()
row
} catch {
case e: Exception =>
kevins-29 marked this conversation as resolved.
Show resolved Hide resolved
if (ins != null) {
finished = true
ins.close()
}
throw e
}
} else {
finished = true
null
}
kevins-29 marked this conversation as resolved.
Show resolved Hide resolved
}
override def close(): Unit = ins.close()
}
}

Expand Down