This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 013ad01887e0 [SPARK-59886][UDF][TEST][FOLLOWUP] Test scalar external
UDFs end to end through a direct gRPC worker
013ad01887e0 is described below
commit 013ad01887e0f2071c29b6954581a1b332bf5a03
Author: Yossi Eliaz <[email protected]>
AuthorDate: Tue Oct 27 15:14:46 2026 +0800
[SPARK-59886][UDF][TEST][FOLLOWUP] Test scalar external UDFs end to end
through a direct gRPC worker
The branch is rebased on current master. Please review the diff at
`a5ad47678e9`.
### What changes were proposed in this pull request?
`PlanExternalUDFsSuite` runs a scalar external UDF through the real
`DIRECT` path. `ExecuteExternalUDFExec` asks `SparkEnv` for the built-in
dispatcher. `DirectGrpcDispatcher` starts `EchoGrpcWorkerMain` and exchanges
Arrow batches with it over a Unix domain socket. The worker returns each batch
unchanged, so the bigint column behaves as an identity function.
The input is `Range(0, 6)` with two partitions. The collected rows are `(0,
0)` through `(5, 5)`. The test runs when
`EchoGrpcWorkerMain.udsTransportAvailable` is true.
`sql/core` depends on `spark-udf-worker-grpc` and its test jar at test
scope, so this suite can load `DirectDispatcherFactory`. `core` does not depend
on gRPC. `SparkEnvUDFDispatcherSuite` covers the missing-runtime error on the
`core` classpath.
`UnixDomainSocketTransport` reports a clear error when neither the epoll
nor the kqueue native transport is available. In
`dev/sparktestsupport/modules.py`, `sql` depends on `udf_worker`, so a change
under `udf/worker/` selects the SQL tests.
`snowflake-jdbc-thin` stays at 4.1.0 from master. The earlier 4.4.0 bump
and StAX service-file strip are gone.
### Why are the changes needed?
The SQL execution node can stream Arrow batches through the built-in direct
dispatcher. This test checks that a query returns rows from the spawned worker.
### Does this PR introduce _any_ user-facing change?
No. The gRPC dependency is test scope. A packaged Spark still reports that
the direct dispatcher runtime is missing.
### How was this patch tested?
`sql/testOnly
org.apache.spark.sql.execution.externalUDF.PlanExternalUDFsSuite -- -z "scalar
external UDF returns rows from a direct gRPC worker"` passed on macOS with JDK
17 after this rebase onto `000f342f737`. Linux CI is [run
39647772434](https://github.com/zozo123/spark/actions/runs/39647772434) on
`a5ad47678e9`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Cursor Grok 4.7
Made with [Cursor](https://cursor.com)
Closes #59155 from zozo123/SPARK-59886-e2e.
Authored-by: Yossi Eliaz <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
dev/sparktestsupport/modules.py | 2 +-
sql/core/pom.xml | 13 ++++++
.../externalUDF/PlanExternalUDFsSuite.scala | 50 ++++++++++++++--------
.../worker/grpc/UnixDomainSocketTransport.scala | 15 ++++++-
.../worker/grpc/testing/EchoGrpcWorkerMain.scala | 13 ++++++
5 files changed, 73 insertions(+), 20 deletions(-)
diff --git a/dev/sparktestsupport/modules.py b/dev/sparktestsupport/modules.py
index 24a5348e6c9e..9f91889310ac 100644
--- a/dev/sparktestsupport/modules.py
+++ b/dev/sparktestsupport/modules.py
@@ -371,7 +371,7 @@ catalyst = Module(
sql = Module(
name="sql",
- dependencies=[catalyst],
+ dependencies=[catalyst, udf_worker],
source_file_regexes=[
"sql/core/",
"python/pyspark/sql/worker/", # analyze_udtf is invoked and tested in
JVM
diff --git a/sql/core/pom.xml b/sql/core/pom.xml
index e11c6d57a15f..6cd71b796c9c 100644
--- a/sql/core/pom.xml
+++ b/sql/core/pom.xml
@@ -330,6 +330,19 @@
<artifactId>arrow-c-data</artifactId>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-udf-worker-grpc_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-udf-worker-grpc_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
index f9ef2a8253d0..b964e27c9c3d 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PlanExternalUDFsSuite.scala
@@ -18,8 +18,9 @@
package org.apache.spark.sql.execution.externalUDF
import java.nio.charset.StandardCharsets
+import java.nio.file.Paths
-import org.apache.spark.{SparkConf, SparkException,
SparkUnsupportedOperationException}
+import org.apache.spark.{SparkConf, SparkUnsupportedOperationException}
import org.apache.spark.sql.{AnalysisException, QueryTest}
import org.apache.spark.sql.catalyst.QueryPlanningTracker
import org.apache.spark.sql.catalyst.expressions.{Add, Alias, And,
ArrayTransform, Attribute,
@@ -34,10 +35,12 @@ import
org.apache.spark.sql.catalyst.plans.logical.{Aggregate, BinaryNode, Execu
Range, Window}
import org.apache.spark.sql.internal.{SQLConf, StaticSQLConf}
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{BooleanType, DataType, IntegerType,
StructField, StructType}
-import org.apache.spark.udf.worker.{DirectWorker, ProcessCallable,
UDFWorkerProperties,
- UDFWorkerSpecification, WorkerEnvironment}
-import org.apache.spark.util.Utils
+import org.apache.spark.sql.types.{BooleanType, DataType, IntegerType,
LongType, StructField,
+ StructType}
+import org.apache.spark.udf.worker.{DirectWorker, ProcessCallable,
UDFProtoCommunicationPattern,
+ UDFWorkerDataFormat, UDFWorkerProperties, UDFWorkerSpecification,
UnixDomainSocket,
+ WorkerCapabilities, WorkerConnectionSpec, WorkerEnvironment}
+import org.apache.spark.udf.worker.grpc.testing.EchoGrpcWorkerMain
class PlanExternalUDFsSuite extends QueryTest with SharedSparkSession {
@@ -716,11 +719,29 @@ class PlanExternalUDFsSuite extends QueryTest with
SharedSparkSession {
assert(execution.workerSpec == spec)
}
- test("scalar external UDF execution reaches dispatcher creation") {
- val spec = workerSpec("unimplemented-scalar")
- val range = Range(0, 10, 1, 1)
- val function = udf("unimplemented-scalar", spec, Seq(singleOutput(range)))
- val resultAttr = AttributeReference("externalUDF", IntegerType, nullable =
true)()
+ test("scalar external UDF returns rows from a direct gRPC worker process") {
+ assume(EchoGrpcWorkerMain.udsTransportAvailable,
+ "Netty UDS native transport (epoll on Linux or kqueue on macOS) is
required")
+ val runner = ProcessCallable.newBuilder()
+ .addCommand(Paths.get(System.getProperty("java.home"), "bin",
"java").toString)
+ .addCommand("-cp")
+ .addCommand(System.getProperty("java.class.path"))
+ .addCommand(classOf[EchoGrpcWorkerMain.type].getName.stripSuffix("$"))
+ val spec = UDFWorkerSpecification.newBuilder()
+ .setEnvironment(WorkerEnvironment.newBuilder())
+ .setCapabilities(WorkerCapabilities.newBuilder()
+ .addSupportedDataFormats(UDFWorkerDataFormat.ARROW)
+
.addSupportedCommunicationPatterns(UDFProtoCommunicationPattern.BIDIRECTIONAL_STREAMING))
+ .setDirect(DirectWorker.newBuilder()
+ .setRunner(runner)
+ .setProperties(UDFWorkerProperties.newBuilder()
+ .setConnection(WorkerConnectionSpec.newBuilder()
+ .setUnixDomainSocket(UnixDomainSocket.getDefaultInstance))
+ .setInitializationTimeoutMs(30000)))
+ .build()
+ val range = Range(0, 6, 1, 2)
+ val function = udf("identity", spec, Seq(singleOutput(range)), dataType =
LongType)
+ val resultAttr = AttributeReference("externalUDF", LongType, nullable =
true)()
val logicalPlan = ExecuteExternalUDF(function, resultAttr, range)
val execution = spark.sessionState.planner.plan(logicalPlan).toSeq match {
case Seq(node: ExecuteExternalUDFExec) => node
@@ -728,13 +749,8 @@ class PlanExternalUDFsSuite extends QueryTest with
SharedSparkSession {
fail(s"Expected one ExecuteExternalUDFExec node,
found:\n${other.mkString("\n")}")
}
- val exception = intercept[SparkException] {
- execution.execute().collect()
- }
- val message = Utils.exceptionString(exception)
- assert(message.contains("built-in dispatcher implementation"))
- assert(message.contains("DIRECT"))
- assert(message.contains("spark-udf-worker-grpc"))
+ val rows = execution.executeCollect().map(row => (row.getLong(0),
row.getLong(1)))
+ assert(rows.toSeq == Seq((0L, 0L), (1L, 1L), (2L, 2L), (3L, 3L), (4L, 4L),
(5L, 5L)))
}
test("map partitions external UDF execution reports unimplemented before
worker startup") {
diff --git
a/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/UnixDomainSocketTransport.scala
b/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/UnixDomainSocketTransport.scala
index 6057ec8482db..6832d7a132dc 100644
---
a/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/UnixDomainSocketTransport.scala
+++
b/udf/worker/grpc/src/main/scala/org/apache/spark/udf/worker/grpc/UnixDomainSocketTransport.scala
@@ -131,8 +131,8 @@ private[grpc] object UnixDomainSocketTransport {
// transport is present the only alternative is a different transport
// entirely -- e.g. TCP loopback -- which is a deployment choice, not a
// fallback we silently substitute here, so we fail loudly instead.
- val epollErr =
Option(epollUnavailable).map(_.getMessage).getOrElse("unknown")
- val kqueueErr =
Option(kqueueUnavailable).map(_.getMessage).getOrElse("unknown")
+ val epollErr = nativeFailure(epollUnavailable)
+ val kqueueErr = nativeFailure(kqueueUnavailable)
throw new UnsupportedOperationException(
s"No Netty native UDS transport available on os=$os " +
s"(epoll: $epollErr, kqueue: $kqueueErr). " +
@@ -141,6 +141,17 @@ private[grpc] object UnixDomainSocketTransport {
}
}
+ private def nativeFailure(cause: Throwable): String = {
+ if (cause == null) {
+ "unknown"
+ } else {
+ Iterator.iterate(cause)(_.getCause).takeWhile(_ != null).take(6).map {
err =>
+ val message = err.getMessage
+ if (message == null || message.isEmpty) err.getClass.getName else
message
+ }.mkString(" | ")
+ }
+ }
+
/**
* Shut an event loop group down. Best-effort: preserves the interrupt
* flag on `InterruptedException` (does not propagate the exception) so
diff --git
a/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala
b/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala
index b48eb57f361c..46a10767c1aa 100644
---
a/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala
+++
b/udf/worker/grpc/src/test/scala/org/apache/spark/udf/worker/grpc/testing/EchoGrpcWorkerMain.scala
@@ -143,6 +143,19 @@ object EchoGrpcWorkerMain {
}
// scalastyle:on println
+ /**
+ * Whether [[UnixDomainSocketTransport.detect]] can select epoll or kqueue.
+ * Callers outside `org.apache.spark.udf.worker.grpc` cannot see that
+ * transport. They assume on this and cancel on hosts where
+ * `DirectGrpcDispatcherIntegrationSuite` cancels.
+ */
+ def udsTransportAvailable: Boolean = try {
+ UnixDomainSocketTransport.detect()
+ true
+ } catch {
+ case _: UnsupportedOperationException => false
+ }
+
/**
* Parses `--key value` flags. The engine guarantees `--id` and
* `--connection` are present; unknown flags are ignored so future
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
