HyukjinKwon commented on code in PR #59934:
URL: https://github.com/apache/spark/pull/59934#discussion_r4465717020
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -970,21 +1192,38 @@ class NativeDataWriter(
/** Creates the `SparkDsTaskContext` of the running task. */
private[ffi] object NativeTask {
/**
- * @param partitionId the partition that the task reads or writes, by
default the partition of
- * the task
+ * @param partitionId the `partition_id` of the context: the index of the
partition in the scan
+ * for a read, and by default the partition of the task
* @param taskAttemptId the ID of the task attempt, by default the one of
the task
+ * @param read whether the context is the one of a read, which the library
may use after the
+ * task ended: an engine may read the stream of a partition
then, and a thread that
+ * the task does not wait for may be in get_next once it
completes, as with
+ * `PipedRDD`. `is_interrupted` then also reports that the task
failed or completed,
+ * and the context does not keep the task alive. A data writer
only runs within its
+ * task, whose kill `is_interrupted` reports. Outside of a task,
it reports nothing.
*/
def create(
partitionId: Option[Int] = None,
- taskAttemptId: Option[Long] = None): NativeLibrary.Task = {
+ taskAttemptId: Option[Long] = None,
+ read: Boolean = false): NativeLibrary.Task = {
val context = Option(TaskContext.get())
+ val interrupted: BooleanSupplier = context match {
+ case Some(taskContext) if read =>
+ val task = new WeakReference(taskContext)
+ () => Option(task.get()).forall(t => t.isInterrupted() || t.isFailed()
|| t.isCompleted())
Review Comment:
Done in d93a1ab: the context of a read registers a failure listener and a
completion listener on the task, which set a flag, so that `is_interrupted`
only reads volatile fields: that flag, and `isInterrupted()` of the task
through the weak reference. The listeners only hold the flag, so the context
still does not keep the task alive. Between the moment the task is marked
failed or completed and the run of the listener, `is_interrupted` can still
return false, which only makes a library that waits stop a little later.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1143,6 +1426,392 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
.mode("append")
.save()
check(writeTasks)
+
+ // A read gets the index of its partition in the scan, also when a task
reads several
+ // partitions, or the partitions of several scans: the task attempt that
read each partition.
+ def readers(dir: File): Map[Int, Long] = dir.listFiles().map { file =>
+ file.getName.toInt -> Files.readString(file.toPath).split(",")(5).toLong
+ }.toMap
+ val (first, second) = (Utils.createTempDir(), Utils.createTempDir())
+ checkRows(
+ range(session, "end" -> "6", "task_path" -> first.getPath).select("id")
+ .union(range(session, "end" -> "6", "task_path" ->
second.getPath).select("id")),
+ (0L until 6L).flatMap(i => Seq(Row(i), Row(i))))
+ assert(readers(first).keySet == Set(0, 1) && readers(second).keySet ==
Set(0, 1))
+ val coalesced = Utils.createTempDir()
+ checkRows(range(session, "end" -> "6", "task_path" ->
coalesced.getPath).select("id")
+ .coalesce(1), (0L until 6L).map(Row(_)))
+ assert(readers(coalesced).keySet == Set(0, 1) &&
readers(coalesced).values.toSet.size == 1)
+ // So does a read of a micro-batch.
+ val (firstStream, secondStream) = (Utils.createTempDir(),
Utils.createTempDir())
+ def counter(dir: File): DataFrame =
session.readStream.format("native_counter")
+ .option("max_offset", "4").option("task_path", dir.getPath).load()
+ val query =
counter(firstStream).union(counter(secondStream)).writeStream.format("noop").start()
+ try query.processAllAvailable() finally query.stop()
+ assert(readers(firstStream).keySet == Set(0) &&
readers(secondStream).keySet == Set(0))
+ }
+
+ nativeTest("open the partitions of batch and micro-batch scans as Arrow C
streams") {
+ val session = newSession(defaultPackage)
+ // The reader factory and the partitions of the batch scan of a query,
which an engine gets
+ // from its plan.
+ def plannedScan(df: DataFrame): (SupportsArrowStreamRead,
Seq[InputPartition]) = {
+ val scan = df.queryExecution.executedPlan.collectFirst { case s:
BatchScanExec => s }.get
+ (scan.readerFactory.asInstanceOf[SupportsArrowStreamRead],
scan.inputPartitions)
+ }
+ // Runs `f` in a task that only gets the files of the session. It is the
task of the partition
+ // 1, so that the partition of the scan that a read gets is not the
partition of its task.
+ def inTask[T: ClassTag](f: => T): T =
session.artifactManager.withResources {
+ session.sparkContext.parallelize(Seq(0, 1), 2).flatMap(i => if (i == 1)
Seq(f) else Nil)
+ .collect().head
+ }
+ import TestArrowEngine.{callbacks, readIds, readIdsInPlace}
+ // Reads the given partitions in a single task, as an engine does.
+ def readAll(factory: SupportsArrowStreamRead, partitions:
Seq[InputPartition]): Seq[Long] = {
+ inTask {
+ partitions.flatMap { partition =>
+ val allocator = ArrowUtils.rootAllocator.newChildAllocator("engine",
0, Long.MaxValue)
+ try {
+ val stream = ArrowArrayStream.allocateNew(allocator)
+ try factory.openArrowStream(partition, stream.memoryAddress())
catch {
+ case e: Throwable =>
+ stream.close()
+ throw e
+ }
+ readIds(stream, allocator)
+ } finally {
+ allocator.close()
+ }
+ }
+ }
+ }
+ // Checks that `f` opens `delta` more task contexts than it closes. The
count is global to the
+ // JVM, where the tasks of earlier tests may still close their contexts,
so it is read right
+ // before `f`.
+ def checkContexts[T](delta: Int)(f: => T): T = {
+ val before = NativeLibrary.openTaskContexts()
+ val result = f
+ assert(NativeLibrary.openTaskContexts() == before + delta)
+ result
+ }
+ // The size of struct ArrowArrayStream, which the layout test compares
with the header.
+ val streamSize = NativeLibrary.layout().get("ArrowArrayStream").longValue
+
+ // An engine reads both partitions of the scan in a single task.
+ val taskDir = Utils.createTempDir()
+ val (factory, partitions) =
+ plannedScan(range(session, "end" -> "6", "task_path" ->
taskDir.getPath).select("id"))
+ assert(partitions.length == 2)
+ assert(readAll(factory, partitions).sorted == (0L until 6L))
+ // The library got the partition of the scan that it reads, not the one of
the task.
+ val tasks = taskDir.listFiles().map { file =>
+ file.getName.toInt ->
Files.readString(file.toPath).split(",").map(_.toLong).toSeq
+ }.toMap
+ assert(tasks.keySet == Set(0, 1) && tasks.values.map(_(5)).toSet.size ==
1, tasks)
+ assert(NativeLibrary.openEngineStreams() == 0)
+
+ // The engine may release the stream after its task ended, and on another
thread: Spark frees
+ // the context of the task once the library released the stream.
+ val released = new File(Utils.createTempDir(), "released")
+ val (laterFactory, laterPartitions) =
+ plannedScan(range(session, "end" -> "6", "released_path" ->
released.getPath).select("id"))
+ // Opens a stream in a task, and returns it unreleased, with what `read`
returned from it in
+ // the task.
+ def openInTask[T: ClassTag](factory: SupportsArrowStreamRead, partition:
InputPartition)(
+ read: ArrowArrayStream => T): (Long, T) = inTask {
+ val stream = Platform.allocateMemory(streamSize)
+ Platform.setMemory(stream, 0.toByte, streamSize)
+ factory.openArrowStream(partition, stream)
+ (stream, read(ArrowArrayStream.wrap(stream)))
+ }
+ val (address, copy) = checkContexts(1) {
+ // The engine reads the stream to its end in the task, but releases it
later.
+ val (address, ids) = openInTask(laterFactory,
laterPartitions.head)(readIdsInPlace)
+ assert(ids == (0L until 3L))
+ assert(NativeLibrary.openEngineStreams() == 1 && !released.exists())
+ // A copy of the stream, as an engine that moves the stream may leave
behind.
+ val copy = Platform.allocateMemory(streamSize)
+ Platform.copyMemory(null, address, null, copy, streamSize)
+ (address, copy)
+ }
+ checkContexts(-1) {
+ ArrowArrayStream.wrap(address).release()
+ assert(NativeLibrary.openEngineStreams() == 0 && released.exists())
+ }
+ Platform.freeMemory(address)
+ // Once the stream is released, the callbacks of the copy fail with
EINVAL, and its release
+ // only marks it released.
+ checkContexts(0) {
+ val array = ArrowArray.allocateNew(ArrowUtils.rootAllocator)
+ val stale = try
intercept[Exception](ArrowArrayStream.wrap(copy).getNext(array)) finally {
+ array.close()
+ }
+ assert(Utils.exceptionString(stale)
+ .contains("The stream was released, or is not a stream of Spark."))
+ ArrowArrayStream.wrap(copy).release()
+ assert(ArrowArrayStream.wrap(copy).snapshot().release == 0)
+ assert(NativeLibrary.openEngineStreams() == 0)
+ }
+ Platform.freeMemory(copy)
+
+ // Once the task ended, its context tells the library that it was
interrupted, so that the
+ // library stops waiting.
+ val marker = new File(Utils.createTempDir(), "interrupted")
+ val (waitingFactory, waitingPartitions) = plannedScan(
+ range(session, "end" -> "6", "wait_for_interrupt" ->
marker.getPath).select("id"))
+ checkContexts(0) {
+ val (waiting, _) = openInTask(waitingFactory, waitingPartitions.head)(_
=> ())
+ val error = intercept[Exception] {
+ readIds(ArrowArrayStream.wrap(waiting), ArrowUtils.rootAllocator)
+ }
+ assert(Utils.exceptionString(error).contains("the task was killed"))
+ assert(Files.readString(marker.toPath) == "interrupted")
+ assert(NativeLibrary.openEngineStreams() == 0)
+ Platform.freeMemory(waiting)
+ }
+
+ // So does the context of a stream that an engine reads in its task once
the task is killed,
+ // although the task did not end.
+ val killedMarker = new File(Utils.createTempDir(), "interrupted")
+ val (killedFactory, killedPartitions) = plannedScan(range(session, "end"
-> "6",
+ "partitions" -> "1", "wait_for_interrupt" ->
killedMarker.getPath).select("id"))
+ val contextsBeforeKill = NativeLibrary.openTaskContexts()
+ val jobGroup = s"native-data-source-${UUID.randomUUID()}"
+ val killError = new AtomicReference[Throwable]()
+ val engineThread = new Thread(() => {
+ session.sparkContext.setJobGroup(jobGroup, "an engine that waits",
interruptOnCancel = true)
+ try {
+ inTask {
+ // Reads the stream on the thread of the task, which waits in
get_next.
+ val stream = ArrowArrayStream.allocateNew(ArrowUtils.rootAllocator)
+ try {
+ killedFactory.openArrowStream(killedPartitions.head,
stream.memoryAddress())
+ try readIdsInPlace(stream) finally stream.release()
+ } finally {
+ stream.close()
+ }
+ }
+ } catch {
+ case e: Throwable => killError.set(e)
+ }
+ })
+ engineThread.start()
+ eventually(timeout(1.minute)) {
+ assert(new File(killedMarker.getPath + ".waiting").isFile)
+ }
+ session.sparkContext.cancelJobGroup(jobGroup)
+ engineThread.join(TimeUnit.MINUTES.toMillis(1))
+ assert(!engineThread.isAlive && killError.get() != null)
+ eventually(timeout(1.minute)) {
+ assert(killedMarker.isFile && Files.readString(killedMarker.toPath) ==
"interrupted")
+ assert(NativeLibrary.openEngineStreams() == 0)
+ assert(NativeLibrary.openTaskContexts() == contextsBeforeKill)
+ }
+
+ // So does the reader factory of a micro-batch scan, with the state of its
micro-batch.
+ val streamTaskDir = Utils.createTempDir()
+ val query =
session.readStream.format("native_counter").option("max_offset", "4")
+ .option("task_path", streamTaskDir.getPath).load()
+ .writeStream.format("noop").start()
+ val microBatchScan = try {
+ query.processAllAvailable()
+
query.asInstanceOf[StreamingQueryWrapper].streamingQuery.lastExecution.executedPlan
+ .collectFirst { case s: MicroBatchScanExec => s }.get
+ } finally {
+ query.stop()
+ }
+ // Only the reads of the engine record their tasks from now on.
+ streamTaskDir.listFiles().foreach(_.delete())
+ val streamFactory =
microBatchScan.readerFactory.asInstanceOf[SupportsArrowStreamRead]
+ val streamPartitions = microBatchScan.inputPartitions.toSeq
+ checkContexts(0) {
+ assert(readAll(streamFactory, streamPartitions).sorted == (0L until 4L))
+ }
+ assert(NativeLibrary.openEngineStreams() == 0)
+ // The library got the partition of the scan, the only one of the
micro-batch.
+ assert(streamTaskDir.listFiles().map(_.getName.toInt).toSet == Set(0))
+ // The partitions of a micro-batch keep the state that the stream reader
returned with them:
+ // a reader factory that an engine keeps reads them with it, also once
Spark planned the next
+ // micro-batch, and so does the reader of Spark.
+ def microBatchStream(options: (String, String)*): MicroBatchStream = {
+ val relation =
session.readStream.format("native_counter").options(options.toMap).load()
+ .queryExecution.analyzed.collectFirst { case relation:
StreamingRelationV2 => relation }.get
+
relation.table.asInstanceOf[SupportsRead].newScanBuilder(relation.extraOptions)
+ .build().toMicroBatchStream(Utils.createTempDir().getPath)
+ }
+ def plan(stream: MicroBatchStream, start: Int, end: Int):
Seq[InputPartition] = {
+ def offset(i: Int): Offset =
stream.deserializeOffset(s"""{"offset":$i}""")
+ stream.planInputPartitions(offset(start), offset(end)).toSeq
+ }
+ val stream = microBatchStream("max_offset" -> "4")
+ try {
+ val first = plan(stream, 0, 2)
+ val keptFactory = stream.createReaderFactory()
+ val second = plan(stream, 2, 4)
+ // The executors get the state of a micro-batch once, as a broadcast
that its partitions
+ // keep, rather than with the partition of each task.
+ Seq(first, second).foreach { partitions =>
+ val partition = partitions.head.asInstanceOf[NativeMicroBatchPartition]
+ assert(partition.state.isDefined && partition.readerState.nonEmpty)
+ val serialized = new ByteArrayOutputStream()
+ Utils.tryWithResource(new
ObjectOutputStream(serialized))(_.writeObject(partition))
+
assert(!serialized.toByteArray.toSeq.containsSlice(partition.readerState.toSeq))
+ }
+ checkContexts(0) {
+ assert(readAll(keptFactory.asInstanceOf[SupportsArrowStreamRead],
first) == Seq(0L, 1L))
+ assert(readAll(keptFactory.asInstanceOf[SupportsArrowStreamRead],
second) == Seq(2L, 3L))
+ val read = inTask {
+ first.flatMap { partition =>
+ val reader = keptFactory.createColumnarReader(partition)
+ try {
+ val ids = ArrayBuffer.empty[Long]
+ while (reader.next()) {
+ val batch = reader.get()
+ (0 until batch.numRows()).foreach(i => ids +=
batch.column(0).getLong(i))
+ }
+ ids.toSeq
+ } finally {
+ reader.close()
+ }
+ }
+ }
+ assert(read == Seq(0L, 1L))
+ }
+ // A micro-batch without partitions broadcasts no state, although the
stream reader
+ // returned one.
+ val nextBroadcast = session.sparkContext.broadcast(0).id + 1
Review Comment:
Done in d93a1ab: `NativeDataSourceStreamReader.lastPlannedState` is a
`private[ffi]` accessor for tests. The test plans a range with partitions and
then an empty one on a stream reader of `native_counter`, and checks that only
the first one has a state, without any broadcast probe.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1143,6 +1426,392 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
.mode("append")
.save()
check(writeTasks)
+
+ // A read gets the index of its partition in the scan, also when a task
reads several
+ // partitions, or the partitions of several scans: the task attempt that
read each partition.
+ def readers(dir: File): Map[Int, Long] = dir.listFiles().map { file =>
+ file.getName.toInt -> Files.readString(file.toPath).split(",")(5).toLong
+ }.toMap
+ val (first, second) = (Utils.createTempDir(), Utils.createTempDir())
+ checkRows(
+ range(session, "end" -> "6", "task_path" -> first.getPath).select("id")
+ .union(range(session, "end" -> "6", "task_path" ->
second.getPath).select("id")),
+ (0L until 6L).flatMap(i => Seq(Row(i), Row(i))))
+ assert(readers(first).keySet == Set(0, 1) && readers(second).keySet ==
Set(0, 1))
+ val coalesced = Utils.createTempDir()
+ checkRows(range(session, "end" -> "6", "task_path" ->
coalesced.getPath).select("id")
+ .coalesce(1), (0L until 6L).map(Row(_)))
+ assert(readers(coalesced).keySet == Set(0, 1) &&
readers(coalesced).values.toSet.size == 1)
+ // So does a read of a micro-batch.
+ val (firstStream, secondStream) = (Utils.createTempDir(),
Utils.createTempDir())
+ def counter(dir: File): DataFrame =
session.readStream.format("native_counter")
+ .option("max_offset", "4").option("task_path", dir.getPath).load()
+ val query =
counter(firstStream).union(counter(secondStream)).writeStream.format("noop").start()
+ try query.processAllAvailable() finally query.stop()
+ assert(readers(firstStream).keySet == Set(0) &&
readers(secondStream).keySet == Set(0))
+ }
+
+ nativeTest("open the partitions of batch and micro-batch scans as Arrow C
streams") {
+ val session = newSession(defaultPackage)
+ // The reader factory and the partitions of the batch scan of a query,
which an engine gets
+ // from its plan.
+ def plannedScan(df: DataFrame): (SupportsArrowStreamRead,
Seq[InputPartition]) = {
+ val scan = df.queryExecution.executedPlan.collectFirst { case s:
BatchScanExec => s }.get
+ (scan.readerFactory.asInstanceOf[SupportsArrowStreamRead],
scan.inputPartitions)
+ }
+ // Runs `f` in a task that only gets the files of the session. It is the
task of the partition
+ // 1, so that the partition of the scan that a read gets is not the
partition of its task.
+ def inTask[T: ClassTag](f: => T): T =
session.artifactManager.withResources {
+ session.sparkContext.parallelize(Seq(0, 1), 2).flatMap(i => if (i == 1)
Seq(f) else Nil)
+ .collect().head
+ }
+ import TestArrowEngine.{callbacks, readIds, readIdsInPlace}
+ // Reads the given partitions in a single task, as an engine does.
+ def readAll(factory: SupportsArrowStreamRead, partitions:
Seq[InputPartition]): Seq[Long] = {
+ inTask {
+ partitions.flatMap { partition =>
+ val allocator = ArrowUtils.rootAllocator.newChildAllocator("engine",
0, Long.MaxValue)
+ try {
+ val stream = ArrowArrayStream.allocateNew(allocator)
+ try factory.openArrowStream(partition, stream.memoryAddress())
catch {
+ case e: Throwable =>
+ stream.close()
+ throw e
+ }
+ readIds(stream, allocator)
+ } finally {
+ allocator.close()
+ }
+ }
+ }
+ }
+ // Checks that `f` opens `delta` more task contexts than it closes. The
count is global to the
+ // JVM, where the tasks of earlier tests may still close their contexts,
so it is read right
+ // before `f`.
+ def checkContexts[T](delta: Int)(f: => T): T = {
+ val before = NativeLibrary.openTaskContexts()
+ val result = f
+ assert(NativeLibrary.openTaskContexts() == before + delta)
+ result
+ }
+ // The size of struct ArrowArrayStream, which the layout test compares
with the header.
+ val streamSize = NativeLibrary.layout().get("ArrowArrayStream").longValue
+
+ // An engine reads both partitions of the scan in a single task.
+ val taskDir = Utils.createTempDir()
+ val (factory, partitions) =
+ plannedScan(range(session, "end" -> "6", "task_path" ->
taskDir.getPath).select("id"))
+ assert(partitions.length == 2)
+ assert(readAll(factory, partitions).sorted == (0L until 6L))
+ // The library got the partition of the scan that it reads, not the one of
the task.
+ val tasks = taskDir.listFiles().map { file =>
+ file.getName.toInt ->
Files.readString(file.toPath).split(",").map(_.toLong).toSeq
+ }.toMap
+ assert(tasks.keySet == Set(0, 1) && tasks.values.map(_(5)).toSet.size ==
1, tasks)
+ assert(NativeLibrary.openEngineStreams() == 0)
+
+ // The engine may release the stream after its task ended, and on another
thread: Spark frees
+ // the context of the task once the library released the stream.
+ val released = new File(Utils.createTempDir(), "released")
+ val (laterFactory, laterPartitions) =
+ plannedScan(range(session, "end" -> "6", "released_path" ->
released.getPath).select("id"))
+ // Opens a stream in a task, and returns it unreleased, with what `read`
returned from it in
+ // the task.
+ def openInTask[T: ClassTag](factory: SupportsArrowStreamRead, partition:
InputPartition)(
+ read: ArrowArrayStream => T): (Long, T) = inTask {
+ val stream = Platform.allocateMemory(streamSize)
+ Platform.setMemory(stream, 0.toByte, streamSize)
+ factory.openArrowStream(partition, stream)
+ (stream, read(ArrowArrayStream.wrap(stream)))
+ }
+ val (address, copy) = checkContexts(1) {
Review Comment:
Done in d93a1ab: the assertions on the stream and its copy run in
`Utils.tryWithSafeFinally`, whose `finally` releases the stream if it is still
open and frees both, so a failed assertion no longer leaves the stream and the
context of its task open.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/v2/ffi/NativeLibrary.java:
##########
@@ -326,6 +433,9 @@ private static final class Ffm {
DOWNCALL = lookup.findVirtual(linker, "downcallHandle",
MethodType.methodType(MethodHandle.class, segment, descriptor,
options))
.asFixedArity();
+ UNBOUND_DOWNCALL = lookup.findVirtual(linker, "downcallHandle",
Review Comment:
Done in d93a1ab: the lookup of `Linker.downcallHandle(FunctionDescriptor,
Linker.Option...)` moved into the initializer of `EngineStreamCallbacks`, with
the method that binds it (`unboundDowncall`, now
`EngineStreamCallbacks.callback`), so `Ffm` no longer has either.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/v2/ffi/NativeLibrary.java:
##########
@@ -903,6 +1158,234 @@ private static boolean isInterrupted(long task) {
}
}
+ /** The number of task contexts that are not closed yet, for tests. */
+ static int openTaskContexts() {
+ return tasks.size();
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // Streams of engines
+ //
---------------------------------------------------------------------------------------------
+
+ // The errno values that the callbacks of a stream return.
+ private static final int EINVAL = 22;
+ private static final int EIO = 5;
+ // The message of get_last_error for a stream that Spark does not know,
which lives as long as
+ // the JVM.
+ private static final long UNKNOWN_STREAM_ERROR =
Review Comment:
Done in d93a1ab: `UNKNOWN_STREAM_ERROR` is a field of
`EngineStreamCallbacks`, allocated in its initializer, and
`engineStreamGetLastError` reads it from there.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/v2/ffi/NativeLibrary.java:
##########
@@ -903,6 +1158,234 @@ private static boolean isInterrupted(long task) {
}
}
+ /** The number of task contexts that are not closed yet, for tests. */
+ static int openTaskContexts() {
+ return tasks.size();
+ }
+
+ //
---------------------------------------------------------------------------------------------
+ // Streams of engines
+ //
---------------------------------------------------------------------------------------------
+
+ // The errno values that the callbacks of a stream return.
+ private static final int EINVAL = 22;
+ private static final int EIO = 5;
+ // The message of get_last_error for a stream that Spark does not know,
which lives as long as
+ // the JVM.
+ private static final long UNKNOWN_STREAM_ERROR =
+ toCString("The stream was released, or is not a stream of Spark.");
+
+ private static final AtomicLong nextEngineStreamId = new AtomicLong(1);
+ // The streams wrapped for engines that are not released yet, by their
private_data.
Review Comment:
Done in d93a1ab: the comment and `nextEngineStreamId` are next to
`engineStreams` again, after `EngineStreamCallbacks`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
