viirya commented on code in PR #58978:
URL: https://github.com/apache/spark/pull/58978#discussion_r4252716326
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/PythonUDF.scala:
##########
@@ -51,7 +51,8 @@ object PythonUDF {
PythonEvalType.SQL_SCALAR_PANDAS_UDF,
PythonEvalType.SQL_SCALAR_PANDAS_ITER_UDF,
PythonEvalType.SQL_SCALAR_ARROW_UDF,
- PythonEvalType.SQL_SCALAR_ARROW_ITER_UDF
+ PythonEvalType.SQL_SCALAR_ARROW_ITER_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_INPROCESS_UDF
Review Comment:
Done in 4dfcc90: `ConvertToCatalyst` passes `parentIsUdf = false` below an
in-process UDF, as `CollapseProject` already excludes it. "apply transpiles a
UDF wrapped by an in-process Python UDF" in `ConvertToCatalystSuite` checks
your `inproc(t(p(x)))` shape: `t` becomes its Catalyst option, and it fails
without the change.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/python/InProcessPythonRuntimeSuite.scala:
##########
@@ -0,0 +1,641 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.python
+
+import java.util.Collections
+import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger,
AtomicReference}
+
+import org.mockito.Mockito.{mock, when}
+
+import org.apache.spark.{SparkConf, SparkFunSuite,
SparkIllegalArgumentException, TaskContext, TaskKilledException}
+import org.apache.spark.api.plugin.PluginContext
+import org.apache.spark.api.python.{ChainedPythonFunctions, PythonEvalType,
SimplePythonFunction}
+import org.apache.spark.internal.config.Python.{IN_PROCESS_PATH_RULE,
IN_PROCESS_SITE_PACKAGES}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.PythonUDF
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.util.ArrowUtils
+
+class InProcessPythonRuntimeSuite extends SparkFunSuite {
+ import InProcessEvaluatorTestUtils._
+
+ private var runtime: InProcessPythonRuntime.InterpreterSession = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ runtime = new InProcessPythonRuntime.InterpreterSession()
+ }
+
+ override def afterEach(): Unit = {
+ try { runtime.shutdown() } finally { super.afterEach() }
+ }
+
+ test("site-packages config validates JEP include paths") {
+ val conf = new SparkConf(false)
+ assert(conf.get(IN_PROCESS_SITE_PACKAGES).isEmpty)
+ conf.set(IN_PROCESS_SITE_PACKAGES.key, " /opt/venv/lib, /opt/extra ")
+ assert(conf.get(IN_PROCESS_SITE_PACKAGES) == Seq("/opt/venv/lib",
"/opt/extra"))
+ conf.set(IN_PROCESS_SITE_PACKAGES.key, "back\\slash")
+ assert(conf.get(IN_PROCESS_SITE_PACKAGES) == Seq("back\\slash"))
+ Seq("bad'path", "bad\npath", "bad\rpath", "bad\u0000path",
+ "bad" + new String(Character.toChars(0x1f600)), "bad" + 0xd800.toChar,
+ s"bad${java.io.File.pathSeparator}path")
+ .foreach { path =>
+ conf.set(IN_PROCESS_SITE_PACKAGES.key, path)
+ intercept[IllegalArgumentException] {
conf.get(IN_PROCESS_SITE_PACKAGES) }
+ intercept[IllegalArgumentException] {
+
InProcessPythonRuntime.InterpreterConfiguration.interpreterConfig(Seq(path))
+ }
+ }
+ }
+
+ /** Shuts the runtime down, returning the warnings about what it still waits
for. */
+ private def shutdownWarnings(waitMillis: Long): Seq[String] = {
+ val appender = new LogAppender("in-process Python shutdown")
+ withLogAppender(appender) { runtime.shutdown(waitMillis) }
+ appender.loggingEvents.map(_.getMessage.getFormattedMessage).toSeq
+ .filter(_.contains("still stopping"))
+ }
+
+ test("shutdown waits for tasks to release their registrations") {
+ val field = ArrowUtils.toArrowField("result", LongType, true, "UTC")
+ intercept[NullPointerException] {
+ runtime.register("idle", Array.emptyByteArray, field, "3.12", false,
false, false, true)
+ }
+ val releaser = thread {
+ Thread.sleep(300)
+ runtime.release(Seq("idle"))
+ }
+ val start = System.nanoTime()
+ runtime.shutdown(waitMillis = 10000)
+ // The last release finishes the shutdown, and shutdown returns once it
has.
+ assert(runtime.isTerminated && System.nanoTime() - start <
TimeUnit.SECONDS.toNanos(5))
+ releaser.join()
+ }
+
+ test("registration failure frees its temporary native command buffer") {
+ val before = ArrowUtils.rootAllocator.getAllocatedMemory
+ val field = ArrowUtils.toArrowField("result", LongType, true, "UTC")
+ intercept[NullPointerException] {
+ // This session deliberately has no interpreter, so invocation fails
after allocation.
+ runtime.register(
+ "failed", new Array[Byte](1024 * 1024), field, "3.12", false, false,
false, true)
+ }
+ assert(ArrowUtils.rootAllocator.getAllocatedMemory == before)
+ assert(shutdownWarnings(waitMillis = 20).exists(_.contains("tasks still
hold")))
+ assert(!runtime.isTerminated)
+ runtime.release(Seq("failed"))
+ runtime.shutdown()
+ assert(runtime.isTerminated)
+ }
+
+ test("plugin reports invalid sitePackages without the installation
checklist") {
+ val ctx = mock(classOf[PluginContext])
+ when(ctx.conf()).thenReturn(new
SparkConf().set(IN_PROCESS_SITE_PACKAGES.key, "/a'b"))
+ val e = intercept[SparkIllegalArgumentException] {
+ new InProcessPythonExecutorPlugin().init(ctx, Collections.emptyMap())
+ }
+ assert(e.getCondition == "INVALID_CONF_VALUE.REQUIREMENT")
+ assert(e.getMessage.contains(IN_PROCESS_PATH_RULE) &&
!e.getMessage.contains("libjep"))
+ }
+
+ test("task-side calls after shutdown report the shutdown") {
+ runtime.shutdown()
+ val field = ArrowUtils.toArrowField("result", LongType, true, "UTC")
+ Seq(
+ () => runtime.onInterpreterThread(()),
+ () => runtime.register("stopped", Array.emptyByteArray, field, "3.12",
+ false, false, false, true)
+ ).foreach { call =>
+ val e = intercept[IllegalStateException] { call() }
+ assert(e.getMessage.contains("has been stopped"))
+ }
+ }
+
+ test("a session that is still stopping rejects a new context") {
+ runtime.requireRunning()
+ runtime.shutdown()
+ val stopping = intercept[InProcessPythonRuntime.LifecycleException] {
+ runtime.requireRunning()
+ }
+ assert(stopping.getMessage.contains("still stopping"))
+ }
+
+ test("sub-millisecond invocations accumulate in processing metrics") {
+ val metric = new SQLMetric("timing", 0L)
+ val timer = new
InProcessArrowEvalPythonEvaluatorFactory.NanosecondTimer(metric)
+ (1 to 25).foreach(_ => timer.add(100000L))
+ assert(metric.value == 2L)
+ timer.add(500000L)
+ assert(metric.value == 3L)
+ }
+
+ test("unused evaluator iterators do not charge Python total time") {
+ val metrics = allMetrics()
+ val context = TaskContext.empty()
+ class TestEvaluator extends InProcessArrowEvalPythonEvaluatorFactory(
+ Seq.empty, Seq.empty, Seq.empty, 10, 0L, "UTC", false, false, false,
false, true, metrics) {
+ override private[python] def runtimeSession:
InProcessPythonRuntime.InterpreterSession =
+ runtime
+
+ def createUnusedIterator(): Unit = {
+ evaluateBatches(Seq.empty, Array.empty, Iterator.empty, new
StructType, context,
+ InProcessArrowEvalPythonEvaluatorFactory.ReadBack)
+ }
+ }
+ new TestEvaluator().createUnusedIterator()
+ Thread.sleep(20)
+ context.markTaskCompleted(None)
+ assert(metrics("pythonTotalTime").value == 0L)
+ }
+
+ test("evaluators retain the generation captured before consuming any input")
{
+ val metrics = allMetrics()
+ val context = TaskContext.empty()
+ val function = SimplePythonFunction(
+ Seq.empty, Collections.emptyMap[String, String](),
Collections.emptyList[String](),
+ "", "3.12", Collections.emptyList(), null)
+ val udf = PythonUDF("identity", function, LongType, Seq.empty,
+ PythonEvalType.SQL_SCALAR_ARROW_INPROCESS_UDF, true)
+ var lookups = 0
+ class TestEvaluator extends InProcessArrowEvalPythonEvaluatorFactory(
+ Seq.empty, Seq(udf), Seq.empty, 10, 0L, "UTC", false, false, false,
false, true, metrics) {
+ override private[python] def runtimeSession:
InProcessPythonRuntime.InterpreterSession = {
+ lookups += 1
+ runtime
+ }
+
+ def iterator(): Iterator[InternalRow] = evaluateBatches(
+ Seq((ChainedPythonFunctions(Seq(function)), 0L)), Array(Array.empty),
+ Iterator.single(InternalRow.empty), new StructType, context,
+ InProcessArrowEvalPythonEvaluatorFactory.ReadBack)
+ }
+ val iterator = new TestEvaluator().iterator()
+ assert(lookups == 1)
+ runtime.shutdown()
+ runtime = new InProcessPythonRuntime.InterpreterSession()
+ try {
+ val error = intercept[IllegalStateException] { iterator.next() }
+ assert(error.getMessage.contains("has been stopped"))
+ assert(lookups == 1)
+ } finally {
+ context.markTaskCompleted(None)
+ }
+ }
+
+ private class Releases {
+ val taskMemory = new AtomicInteger()
+ val abandoned = new AtomicInteger()
+ val recorded = new AtomicInteger()
+ val others = new AtomicInteger()
+
+ /** Resources whose consumer has started, so that closing records the time
once. */
+ def resources(lockWaitMillis: Long = 10000L)
+ : InProcessArrowEvalPythonEvaluatorFactory.IteratorResources = {
+ val resources = new
InProcessArrowEvalPythonEvaluatorFactory.IteratorResources(
+ () => taskMemory.incrementAndGet(),
+ () => abandoned.incrementAndGet(),
+ _ => recorded.incrementAndGet(),
Review Comment:
Done in 4dfcc90: the test helper keeps the last recorded value, and "task
completion records the time since the consumer started, once" sleeps 50 ms
after `start()`, calls `start()` again and closes twice, then checks one
recording of at least 50 ms and less than 30 s. Recording `startedAt` itself,
or resetting the start time on a later `start()`, fails it.
##########
docs/sql-pyspark-inprocess-udf.md:
##########
@@ -0,0 +1,737 @@
+---
+layout: global
+title: In-Process Python UDFs
+displayTitle: In-Process Python UDFs
+license: |
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+---
+
+* Table of contents
+{:toc}
+
+## Runtime and result contract
+
+Each executor owns a dedicated interpreter thread. The plugin initializes the
+interpreter on that thread, and task calls and shutdown are dispatched to the
+same thread. The JVM is asked to allocate an 8 MiB stack for this thread; the
+actual size is platform-dependent. Calls from concurrent tasks are queued on
the
+interpreter thread.
+One task per executor is recommended for throughput, but is not a correctness
requirement.
+Application-level Python parallelism comes from multiple executor JVMs.
+The plugin configures JEP's process-wide interpreter with hash seed `0`,
matching
+Spark's default Python worker seed. It must initialize before any other JEP
user in
+the JVM. The seed cannot change between SparkContexts in the same process; a
custom
+worker `PYTHONHASHSEED` does not override this embedded-runtime setting.
+
+Task cancellation cannot safely stop arbitrary native Python code. An
interrupted
+caller waits for the current invocation to finish before freeing the Arrow CDI
+structures, then restores its interrupt status. A UDF that never returns can
+therefore prevent its task from completing cancellation and block every
subsequent
+in-process UDF on that executor, including calls from other tasks, jobs, and
sessions.
+Recovery from a permanently hung invocation requires replacing the executor
process.
+Plugin shutdown stops accepting new calls and waits up to five seconds for the
interpreter thread. If a call is
+still running or a task still owns exported results, cleanup waits for that
task to release
+its CDI references; the memory remains live until cleanup completes or the
process exits. Shutdown does not forcibly interrupt native
+code. A new interpreter cannot start until the previous one has fully stopped.
+
+A scalar UDF must return a `pyarrow.Array` with exactly one element per input
row.
+The runtime checks the result type against the declared Spark type, including
+nested fields, decimal scale, and timestamp unit. Timezone-aware timestamps
are relabeled
+to `spark.sql.session.timeZone` without changing their UTC instants or copying
their buffers.
+Timezone-naive and timezone-aware timestamps are not interchangeable. String
and binary
+offset widths, including nested values, are converted as needed to match
+`spark.sql.execution.arrow.useLargeVarTypes`. Large, fixed-size and
dictionary-encoded
+representations of the declared types (`large_list`, `fixed_size_list`,
`string_view`,
+`binary_view`, `fixed_size_binary` and dictionary arrays) are cast to the
declared type.
+These conversions can allocate new buffers. Other value types must match
exactly: use an
+explicit PyArrow cast for numeric conversions.
+Map `keys_sorted` metadata is normalized to Spark's declared map type.
+Nested field nullability may differ if the actual values satisfy the declared
nullability. Sliced results, including nested
+child slices, are copied to remove offsets that Arrow Java's CDI importer
cannot
+read. Zero-length levels without a usable offsets buffer, which Arrow permits,
are given
+one. Compatible results retain zero-copy transfer.
+Before exporting a result, the runtime performs full Arrow validation,
including interior
+offsets, because the JVM reads result buffers without bounds checks: a
malformed result,
+such as one built from raw buffers, could otherwise produce wrong values or
crash the
+executor. It does not validate UTF-8 in string results, because Spark strings
may contain
+invalid UTF-8 (for example, `CAST(X'FF' AS STRING)`). Worker-based Arrow UDFs
do not
+validate their results. To skip the full validation, set
+`spark.sql.execution.pythonUDF.inProcess.fullValidation.enabled` to `false`;
Arrow's
+constant-time validation and the conversions above still apply.
+
+The API produces a regular `PythonUDF` expression with an in-process evaluation
+type. Spark's existing `ArrowEvalPython` planning rules handle aggregation,
Review Comment:
Done in 4dfcc90: the guide says that in-process UDFs cannot be called inside
the lambda of a higher-order function such as `transform` or `filter`, and that
such a call fails analysis with
`UNSUPPORTED_FEATURE.LAMBDA_FUNCTION_WITH_PYTHON_UDF`. "in-process UDFs inside
higher-order function lambdas are rejected" pins that error.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/python/InProcessArrowEvalPythonEvaluatorFactory.scala:
##########
@@ -0,0 +1,579 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.python
+
+import java.io.File
+import java.nio.file.Files
+import java.util.UUID
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
+import java.util.concurrent.locks.ReentrantLock
+
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+import com.google.common.util.concurrent.Uninterruptibles
+import org.apache.arrow.c.{ArrowArray, ArrowSchema, BaseStruct}
+import org.apache.arrow.util.AutoCloseables
+import org.apache.arrow.vector.VectorSchemaRoot
+
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
+import org.apache.spark.api.python.ChainedPythonFunctions
+import org.apache.spark.memory.MemoryConsumer
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression,
JoinedRow, PythonUDF, UnsafeProjection, UnsafeRow}
+import org.apache.spark.sql.execution.arrow.ArrowWriter
+import org.apache.spark.sql.execution.metric.SQLMetric
+import org.apache.spark.sql.execution.python.EvalPythonExec.ArgumentMetadata
+import org.apache.spark.sql.types._
+import org.apache.spark.sql.util.ArrowUtils
+import org.apache.spark.sql.vectorized.{ArrowColumnVector, ColumnarBatch,
ColumnVector}
+import org.apache.spark.util.Utils
+
+/**
+ * Evaluates scalar Python UDFs using Arrow CDI in the executor process. Only
UDF arguments
+ * are converted to Arrow. Original rows are buffered in a spillable queue and
joined with
+ * the results, unless all of them are UDF arguments that read back from Arrow
unchanged.
+ * Each batch owns its Arrow buffers so Python can safely retain input arrays.
+ *
+ * The evaluator owns its queue, so that cleanup at task completion is
coordinated with a
+ * consumer on another thread, such as a pipelined Python writer or a
TRANSFORM feed thread.
+ */
+class InProcessArrowEvalPythonEvaluatorFactory(
+ childOutput: Seq[Attribute],
+ udfs: Seq[PythonUDF],
+ output: Seq[Attribute],
+ batchSize: Int,
+ maxBytes: Long,
+ timeZoneId: String,
+ largeVarTypes: Boolean,
+ hideTraceback: Boolean,
+ simplifiedTraceback: Boolean,
+ tracebackWithLocals: Boolean,
+ fullValidation: Boolean,
+ metrics: Map[String, SQLMetric])
+ extends EvalPythonEvaluatorFactory(childOutput, udfs, output) {
+
+ private[python] def runtimeSession:
InProcessPythonRuntime.InterpreterSession =
+ InProcessPythonRuntime.currentSession
+
+ /** Unused: `evaluateJoined` always evaluates the UDFs. */
+ override protected def evaluate(
+ funcs: Seq[(ChainedPythonFunctions, Long)],
+ argMetas: Array[Array[ArgumentMetadata]],
+ rows: Iterator[InternalRow],
+ inputSchema: StructType,
+ context: TaskContext): Iterator[InternalRow] =
+ throw SparkException.internalError("In-process UDFs are evaluated with
their input rows")
+
+ override protected def evaluateJoined(
+ funcs: Seq[(ChainedPythonFunctions, Long)],
+ argMetas: Array[Array[ArgumentMetadata]],
+ rows: Iterator[InternalRow],
+ inputs: Seq[Expression],
+ inputSchema: StructType,
+ context: TaskContext): Option[Iterator[InternalRow]] = {
+ import InProcessArrowEvalPythonEvaluatorFactory.{Buffered, ReadBack,
readsBack}
+ val inputColumns = inputs.length == childOutput.length &&
inputs.zip(childOutput).forall {
+ case (a: Attribute, c) => a.exprId == c.exprId
+ case _ => false
+ }
+ // If all input columns are UDF arguments, they are written to Arrow
regardless. Read them
+ // back from the exported input vectors instead of buffering every input
row, if their
+ // values read back from Arrow exactly as written and as fast as an unsafe
row copy.
+ val joinInput = if (inputColumns && inputSchema.forall(f =>
readsBack(f.dataType))) {
+ ReadBack
+ } else if (inputColumns) {
+ Buffered(None)
+ } else {
+ // Each projected row is written to Arrow before the next input row is
pulled, so the
+ // arguments go into a reused buffer rather than being copied value by
value.
+ val projection = UnsafeProjection.create(inputs, childOutput)
+ projection.initialize(context.partitionId())
+ Buffered(Some(projection))
+ }
+ Some(evaluateBatches(funcs, argMetas, rows, inputSchema, context,
joinInput))
+ }
+
+ private[python] def evaluateBatches(
+ funcs: Seq[(ChainedPythonFunctions, Long)],
+ argMetas: Array[Array[ArgumentMetadata]],
+ rows: Iterator[InternalRow],
+ inputSchema: StructType,
+ context: TaskContext,
+ joinInput: InProcessArrowEvalPythonEvaluatorFactory.JoinInput):
Iterator[InternalRow] = {
+ import InProcessArrowEvalPythonEvaluatorFactory.{Buffered, ReadBack}
+ ArrowUtils.failDuplicatedFieldNames(inputSchema)
+ val functions = funcs.map { case (chain, _) =>
+ if (chain.funcs.size != 1) {
+ throw SparkException.internalError(
+ "In-process UDF chains must use separate evaluation nodes")
+ }
+ chain.funcs.head
+ }
+ val inputOrdinals = argMetas.map(_.map(_.offset))
+ def checkCancellation(): Unit = context.killTaskIfInterrupted()
+
+ val expectedFields = udfs.map { udf =>
+ ArrowUtils.toArrowField("result", udf.dataType, true, timeZoneId,
largeVarTypes)
+ }
+ val processingTime = new
InProcessArrowEvalPythonEvaluatorFactory.NanosecondTimer(
+ metrics("pythonProcessingTime"))
+ val initTime = new
InProcessArrowEvalPythonEvaluatorFactory.NanosecondTimer(
+ metrics("pythonInitTime"))
+ val arrowSchema = ArrowUtils.toArrowSchema(inputSchema, timeZoneId,
largeVarTypes)
+ // Capture before consuming input: an old task must never join a later
context's session.
+ val runtime = runtimeSession
+ // Rows are copied out of the queue and Arrow vectors before they are
returned, so they
+ // remain valid after task completion releases those, on whichever thread
consumes them.
+ val resultProj = UnsafeProjection.create(output, output)
+ // Spill files go into a directory of the queue's own, created with the
first disk queue, so
+ // that task completion can delete them when it cannot close the queue.
+ @volatile var spillDir: File = null
+ // Guarded by the queue's monitor.
+ var queueAbandoned = false
+ val (queue, projection) = joinInput match {
+ case Buffered(projection) =>
+ val localDir = new File(Utils.getLocalDir(SparkEnv.get.conf))
+ val serializerManager = SparkEnv.get.serializerManager
+ // Only the consumer holding the iterator's lock adds and removes rows.
+ val queue = new HybridRowQueue(context.taskMemoryManager(), localDir,
+ childOutput.length, serializerManager, lockFree = true) {
+ override protected def createDiskQueue(): RowQueue = synchronized {
+ if (spillDir == null) {
+ spillDir = Files.createTempDirectory(localDir.toPath,
"inprocess-udf-").toFile
+ }
+ DiskRowQueue(Files.createTempFile(spillDir.toPath, "buffer",
"").toFile,
+ childOutput.length, serializerManager)
+ }
+
+ // Once task completion leaves the queue to the executor, it must
not spill for other
+ // consumers into a directory that nothing deletes.
+ override def spill(size: Long, trigger: MemoryConsumer): Long =
synchronized {
+ if (queueAbandoned) 0L else super.spill(size, trigger)
+ }
+
+ // Queues of a task are distinct memory consumers, whatever their
case-class fields.
+ override def equals(other: Any): Boolean = this eq
other.asInstanceOf[AnyRef]
+ override def hashCode(): Int = System.identityHashCode(this)
+ override def canEqual(other: Any): Boolean = false
+ }
+ (queue, projection.orNull)
+ case ReadBack => (null, null)
+ }
+ val joined = new JoinedRow
+ val handles = functions.map(_ => UUID.randomUUID().toString)
+ var registered = false
+ var writer: ArrowWriter = null
+ val results = ArrayBuffer.empty[ArrowColumnVector]
+ // Whether the consumer has started, which `resources` records for the
total time.
+ var started = false
Review Comment:
Done in 4dfcc90 as suggested: `IteratorResources.start()` keeps a plain
`started` flag, set by the consumer with the lock held, so that only the first
call sets the volatile start time, and the evaluator calls it without its own
flag.
--
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]
