From cefac2318d7233aad965ae68bac6220d35cd365d Mon Sep 17 00:00:00 2001
From: Haiyang Sun <75672016+haiyangsun-db@users.noreply.github.com>
Date: Wed, 9 Sep 2026 09:09:03 +0000
Subject: [PATCH 1/2] [SPARK-59364][UDF] Build external UDF payload and Init
 messages

---
 .../ExternalUserDefinedFunction.scala         | 205 +++++++-
 .../PythonExternalUserDefinedFunction.scala   |  51 ++
 .../externalUDF/PythonUDFPayload.scala        | 135 ++++++
 .../PythonUDFWorkerSpecification.scala        |  38 +-
 .../execution/python/ArrowPythonRunner.scala  |  35 +-
 ...thonExternalUserDefinedFunctionSuite.scala | 450 ++++++++++++++++++
 .../proto/src/main/protobuf/udf_message.proto |  90 +++-
 .../proto/src/main/protobuf/worker_spec.proto |  60 +++
 8 files changed, 1009 insertions(+), 55 deletions(-)
 create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala
 create mode 100644 sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala
 create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExternalUserDefinedFunction.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExternalUserDefinedFunction.scala
index 0bbaf1ee242db..b6d1202215c6b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExternalUserDefinedFunction.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExternalUserDefinedFunction.scala
@@ -17,13 +17,61 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import scala.jdk.CollectionConverters._
+
+import com.google.protobuf.ByteString
+
 import org.apache.spark.annotation.Experimental
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.analysis.TypeCheckResult.TypeCheckSuccess
 import org.apache.spark.sql.catalyst.trees.TreePattern.{EXTERNAL_UDF, TreePattern}
 import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.types.DataType
-import org.apache.spark.udf.worker.UDFWorkerSpecification
+import org.apache.spark.sql.types.{DataType, StructField, StructType}
+import org.apache.spark.udf.worker.{DynamicConfigRequirement, Init, UdfArgument, UdfInputMetadata,
+  UdfPayload, UDFWorkerDataFormat, UDFWorkerSpecification, WorkerContextReference}
+
+/**
+ * Language-neutral inputs used by an [[ExternalUserDefinedFunction]] to initialize one worker
+ * session. The physical operator owns these engine-side values. [[ExternalUserDefinedFunction]]
+ * combines them with the context declaration in its worker specification.
+ *
+ * The environment, dynamic configuration, and resource-directory maps may contain only
+ * engine-authorized, resolved values. [[ExternalUserDefinedFunction]] selects the names requested
+ * by the worker specification. In particular, callers must not expose the engine's complete
+ * process environment through [[environmentVariables]]. Callers also resolve engine/task state,
+ * including the task-context keys and resource directories required by a worker; the generic
+ * builder only validates and forwards those values. Physical execution must populate
+ * [[ExternalUDFInitContext.DRIVER_ID_CONTEXT_KEY]] and
+ * [[ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY]] for every session. The driver ID defaults to
+ * `SparkContext.DRIVER_IDENTIFIER`; the driver-role flag is derived from the local `SparkEnv`.
+ */
+@Experimental
+case class ExternalUDFInitContext(
+    protocolVersion: Int,
+    dataFormat: UDFWorkerDataFormat,
+    inputSchema: Array[Byte],
+    outputSchema: Array[Byte],
+    timezone: String,
+    taskContext: Map[String, String] = Map.empty,
+    environmentVariables: Map[String, String] = Map.empty,
+    dynamicConfig: Map[String, String] = Map.empty,
+    resourceDirectories: Map[String, String] = Map.empty) {
+
+  private[expressions] def newInitBuilder(): Init.Builder = {
+    Init.newBuilder()
+      .setProtocolVersion(protocolVersion)
+      .setDataFormat(dataFormat)
+      .setInputSchema(ByteString.copyFrom(inputSchema))
+      .setOutputSchema(ByteString.copyFrom(outputSchema))
+      .setTimezone(timezone)
+      .putAllTaskContext(taskContext.asJava)
+  }
+}
+
+object ExternalUDFInitContext {
+  val DRIVER_ID_CONTEXT_KEY: String = "driverId"
+  val IS_DRIVER_CONTEXT_KEY: String = "isDriver"
+}
 
 /**
  * :: Experimental ::
@@ -40,15 +88,21 @@ import org.apache.spark.udf.worker.UDFWorkerSpecification
  * operator (e.g. [[org.apache.spark.sql.execution.externalUDF.MapPartitionsExternalUDFExec]])
  * to execute.
  *
- * @param name             Optional name of the UDF.
- * @param workerSpec       Specification of the worker that executes this UDF.
- * @param payload          Opaque serialized function definition.
- * @param dataType         Return type of the UDF.
- * @param children         Input argument expressions.
- * @param inputTypes       Optional declared input types for validation.
+ * The worker specification declares the engine context needed for each worker session. This
+ * expression supplies the invocation-specific payload metadata. Together they are sufficient to
+ * construct [[Init]] without a language-specific expression subtype.
+ *
+ * @param name Optional name of the UDF.
+ * @param workerSpec Specification of the worker that executes this UDF.
+ * @param payload Opaque serialized function definition.
+ * @param dataType Return type of the UDF.
+ * @param children Input argument expressions.
+ * @param inputTypes Optional declared input types for validation.
  * @param udfDeterministic Whether this UDF is deterministic.
- * @param udfNullable      Whether this UDF can return null.
- * @param resultId         Unique expression ID for this invocation.
+ * @param udfNullable Whether this UDF can return null.
+ * @param resultId Unique expression ID for this invocation.
+ * @param payloadFormat Format tag identifying the opaque payload encoding.
+ * @param evalType Optional worker-specific dispatch hint.
  */
 @Experimental
 case class ExternalUserDefinedFunction(
@@ -60,9 +114,133 @@ case class ExternalUserDefinedFunction(
     inputTypes: Option[Seq[DataType]] = None,
     udfDeterministic: Boolean,
     udfNullable: Boolean,
-    resultId: ExprId = NamedExpression.newExprId)
+    resultId: ExprId = NamedExpression.newExprId,
+    payloadFormat: String = ExternalUserDefinedFunction.DEFAULT_PAYLOAD_FORMAT,
+    evalType: Option[String] = None)
   extends Expression with NonSQLExpression with Unevaluable {
 
+  require(payloadFormat.nonEmpty, "External UDF payload format must be non-empty")
+
+  /** Builds the complete initialization message for one worker session. */
+  def buildInit(context: ExternalUDFInitContext): Init = {
+    val udfBuilder = UdfPayload.newBuilder()
+      .setPayload(ByteString.copyFrom(payload))
+      .setFormat(payloadFormat)
+      .setInvocationId(resultId.id)
+    name.foreach(udfBuilder.setName)
+    evalType.foreach(udfBuilder.setEvalType)
+
+    val arguments = children.map {
+      case NamedArgumentExpression(argumentName, value) => (value, Some(argumentName))
+      case expression => (expression, None)
+    }
+    val inputSchema = StructType(arguments.zipWithIndex.map {
+      case ((expression, _), index) =>
+        StructField(s"_$index", expression.dataType, expression.nullable)
+    })
+    val inputBuilder = UdfInputMetadata.newBuilder()
+      .setSchemaFormat(ExternalUserDefinedFunction.INPUT_SCHEMA_FORMAT)
+      .setSchema(ByteString.copyFromUtf8(inputSchema.json))
+    arguments.zipWithIndex.foreach { case ((_, argumentName), offset) =>
+      val argumentBuilder = UdfArgument.newBuilder().setInputOffset(offset)
+      argumentName.foreach(argumentBuilder.setName)
+      inputBuilder.addArguments(argumentBuilder)
+    }
+    udfBuilder.setInput(inputBuilder)
+
+    val session = workerSpec.getSession
+    val environmentVariables = requestedValues(
+      session.getEnvironmentVariableReferencesMap.asScala.toMap,
+      context.environmentVariables)
+    val staticConfig = session.getStaticConfigMap.asScala.toMap
+    val dynamicConfigRequirements = session.getDynamicConfigMap.asScala.toMap
+    validateUniqueKeys(
+      staticConfig.keySet,
+      dynamicConfigRequirements.keySet)
+    val dynamicConfig = requestedDynamicConfig(
+      dynamicConfigRequirements,
+      context.dynamicConfig)
+    val resourceDirectories = requestedResourceDirectories(
+      session.getRequiredResourceDirectoriesList.asScala.toSeq,
+      context.resourceDirectories)
+
+    val initBuilder = context.newInitBuilder()
+      .setUdf(udfBuilder)
+      .putAllEnvironmentVariables(environmentVariables.asJava)
+      .putAllSessionConf((staticConfig ++ dynamicConfig).asJava)
+      .putAllResourceDirectories(resourceDirectories.asJava)
+    initBuilder.build()
+  }
+
+  private def requestedDynamicConfig(
+      requirements: Map[String, DynamicConfigRequirement],
+      values: Map[String, String]): Map[String, String] = {
+    val missingRequired = requirements.iterator.collect {
+      case (name, requirement) if requirement.getIsRequired && !values.contains(name) => name
+    }.toSeq.sorted
+    require(
+      missingRequired.isEmpty,
+      s"Missing required dynamic configuration: ${missingRequired.mkString(", ")}")
+    requirements.keysIterator.flatMap { name =>
+      values.get(name).map { value =>
+        require(value != null, s"Null dynamic configuration value for $name")
+        name -> value
+      }
+    }.toMap
+  }
+
+  private def requestedResourceDirectories(
+      requiredNames: Seq[String],
+      values: Map[String, String]): Map[String, String] = {
+    require(
+      !requiredNames.contains(""),
+      "Worker session contains an empty required resource directory name")
+    val duplicateNames = requiredNames.groupBy(identity)
+      .collect { case (name, occurrences) if occurrences.size > 1 => name }
+      .toSeq
+      .sorted
+    require(
+      duplicateNames.isEmpty,
+      s"Worker session contains duplicate required resource directories: " +
+        duplicateNames.mkString(", "))
+    val missingNames = requiredNames.filterNot(values.contains).sorted
+    require(
+      missingNames.isEmpty,
+      s"Missing required resource directories: ${missingNames.mkString(", ")}")
+    val emptyNames = requiredNames.filter { name =>
+      val directory = values(name)
+      directory == null || directory.isEmpty
+    }.sorted
+    require(
+      emptyNames.isEmpty,
+      s"Empty required resource directories: ${emptyNames.mkString(", ")}")
+    requiredNames.iterator.map(name => name -> values(name)).toMap
+  }
+
+  private def requestedValues(
+      references: Map[String, WorkerContextReference],
+      values: Map[String, String]): Map[String, String] = {
+    references.iterator.flatMap { case (target, reference) =>
+      require(target.nonEmpty, "Empty worker context target")
+      require(reference.getSource.nonEmpty, s"Empty worker context source for $target")
+      values.get(reference.getSource)
+        .orElse(if (reference.hasDefaultValue) Some(reference.getDefaultValue) else None)
+        .map(target -> _)
+    }.toMap
+  }
+
+  private def validateUniqueKeys(keySets: Set[String]*): Unit = {
+    require(!keySets.exists(_.contains("")), "Worker session context contains an empty target key")
+    val duplicateKeys = keySets.iterator.flatten.toSeq
+      .groupBy(identity)
+      .collect { case (key, occurrences) if occurrences.size > 1 => key }
+      .toSeq
+      .sorted
+    require(
+      duplicateKeys.isEmpty,
+      s"Worker session context contains duplicate target keys: ${duplicateKeys.mkString(", ")}")
+  }
+
   override lazy val deterministic: Boolean = udfDeterministic && children.forall(_.deterministic)
 
   override def nullable: Boolean = udfNullable
@@ -98,3 +276,8 @@ case class ExternalUserDefinedFunction(
       newChildren: IndexedSeq[Expression]): ExternalUserDefinedFunction =
     copy(children = newChildren)
 }
+
+object ExternalUserDefinedFunction {
+  val DEFAULT_PAYLOAD_FORMAT: String = "raw-v1"
+  val INPUT_SCHEMA_FORMAT: String = "spark-sql-data-type-json-v1"
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala
new file mode 100644
index 0000000000000..f9791ffbabcb5
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala
@@ -0,0 +1,51 @@
+/*
+ * 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.externalUDF
+
+import org.apache.spark.api.python.PythonEvalType
+import org.apache.spark.sql.catalyst.expressions.{ExternalUserDefinedFunction, PythonUDF}
+import org.apache.spark.udf.worker.UDFWorkerSpecification
+
+/** Converts PySpark UDF metadata to the language-neutral external UDF representation. */
+private[sql] object PythonExternalUserDefinedFunction {
+  val PAYLOAD_FORMAT: String = "pyspark-udf-v2"
+
+  /**
+   * Creates a scalar Python external UDF while leaving worker launch policy to the caller. Python
+   * metadata stays inside the opaque payload; Init construction remains language-independent.
+   */
+  def fromPythonUDF(
+      udf: PythonUDF,
+      workerSpec: UDFWorkerSpecification): ExternalUserDefinedFunction = {
+    require(
+      udf.evalType == PythonEvalType.SQL_ARROW_BATCHED_UDF,
+      s"Unsupported Python external UDF eval type: ${udf.evalType}")
+
+    ExternalUserDefinedFunction(
+      name = Option(udf.name),
+      workerSpec = workerSpec,
+      payload = PythonUDFPayload.encode(udf.func),
+      dataType = udf.dataType,
+      children = udf.children,
+      inputTypes = None,
+      udfDeterministic = udf.udfDeterministic,
+      udfNullable = udf.nullable,
+      resultId = udf.resultId,
+      payloadFormat = PAYLOAD_FORMAT,
+      evalType = Some(udf.evalType.toString))
+  }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala
new file mode 100644
index 0000000000000..6c70de203ddea
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala
@@ -0,0 +1,135 @@
+/*
+ * 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.externalUDF
+
+import java.io.{ByteArrayInputStream, ByteArrayOutputStream, DataInputStream, DataOutputStream,
+  EOFException}
+import java.nio.charset.StandardCharsets.UTF_8
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.api.python.PythonFunction
+
+/**
+ * Versioned static payload for a scalar PySpark UDF.
+ *
+ * Only Python-private state which does not change when Catalyst rewrites an invocation is stored
+ * here. Logical input types, argument bindings, resource locations, and runtime state are supplied
+ * by language-neutral Init fields.
+ */
+private[sql] final class PythonUDFPayload private (
+    commandBytes: Array[Byte],
+    val pythonIncludes: Vector[String],
+    val pythonVersion: String) {
+
+  def command: Array[Byte] = commandBytes.clone()
+}
+
+private[sql] object PythonUDFPayload {
+  // ASCII "PYUD" followed by a format version.
+  private val MAGIC = 0x50595544
+  private val VERSION = 2
+
+  /** Encodes all Python-specific state which is invariant across task attempts. */
+  def encode(func: PythonFunction): Array[Byte] = {
+    // TODO(SPARK-59366): Carry `func.broadcastVars` through language-neutral UDF resource
+    // metadata before routing PySpark scalar UDFs through unified execution.
+    val pythonIncludes = Option(func.pythonIncludes)
+      .map(_.asScala.toVector)
+      .getOrElse(Vector.empty)
+    encodeFields(
+      func.command.toArray,
+      pythonIncludes,
+      func.pythonVer)
+  }
+
+  private def encodeFields(
+      command: Array[Byte],
+      pythonIncludes: Seq[String],
+      pythonVersion: String): Array[Byte] = {
+    val buffer = new ByteArrayOutputStream()
+    val output = new DataOutputStream(buffer)
+
+    output.writeInt(MAGIC)
+    output.writeInt(VERSION)
+    writeBytes(command, output)
+    output.writeInt(pythonIncludes.size)
+    pythonIncludes.foreach { pythonInclude =>
+      require(pythonInclude != null, "Python include cannot be null")
+      writeString(pythonInclude, output)
+    }
+    writeString(pythonVersion, output)
+    output.flush()
+    buffer.toByteArray
+  }
+
+  def decode(payload: Array[Byte]): PythonUDFPayload = {
+    require(payload != null, "Python UDF payload cannot be null")
+    val input = new DataInputStream(new ByteArrayInputStream(payload))
+    try {
+      requireField(input.readInt() == MAGIC, "invalid magic")
+      val version = input.readInt()
+      requireField(version == VERSION, s"unsupported version $version")
+
+      val command = readBytes(input, "command")
+      val includeCount = input.readInt()
+      requireField(
+        includeCount >= 0 && includeCount <= input.available() / Integer.BYTES,
+        s"invalid Python include count $includeCount")
+      val pythonIncludes = Vector.fill(includeCount) {
+        readString(input, "Python include")
+      }
+      val pythonVersion = readString(input, "Python version")
+      requireField(input.available() == 0, "trailing bytes")
+      new PythonUDFPayload(
+        command,
+        pythonIncludes,
+        pythonVersion)
+    } catch {
+      case e: EOFException =>
+        throw new IllegalArgumentException("Malformed Python UDF payload", e)
+    }
+  }
+
+  private def writeBytes(bytes: Array[Byte], output: DataOutputStream): Unit = {
+    output.writeInt(bytes.length)
+    output.write(bytes)
+  }
+
+  private def writeString(value: String, output: DataOutputStream): Unit = {
+    require(value != null, "Python UDF payload string cannot be null")
+    writeBytes(value.getBytes(UTF_8), output)
+  }
+
+  private def readBytes(input: DataInputStream, field: String): Array[Byte] = {
+    val length = input.readInt()
+    requireField(length >= 0 && length <= input.available(), s"invalid $field length $length")
+    val bytes = new Array[Byte](length)
+    input.readFully(bytes)
+    bytes
+  }
+
+  private def readString(input: DataInputStream, field: String): String = {
+    new String(readBytes(input, field), UTF_8)
+  }
+
+  private def requireField(condition: Boolean, detail: => String): Unit = {
+    if (!condition) {
+      throw new IllegalArgumentException(s"Malformed Python UDF payload: $detail")
+    }
+  }
+}
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala
index aa5a0850580ae..8a39d11c01fa4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala
@@ -21,7 +21,10 @@ import scala.jdk.CollectionConverters._
 import org.apache.spark.SparkConf
 import org.apache.spark.annotation.Experimental
 import org.apache.spark.api.python.{PythonFunction, PythonUtils}
+import org.apache.spark.internal.config.OptionalConfigEntry
 import org.apache.spark.internal.config.Python.PYTHON_WORKER_MODULE
+import org.apache.spark.sql.execution.python.ArrowPythonRunner
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.udf.worker._
 
 /**
@@ -37,14 +40,15 @@ import org.apache.spark.udf.worker._
  *    Spark's built-in Python path and the system `PYTHONPATH`
  *  - Worker module from `spark.python.worker.module`
  *
- * Note: `pythonIncludes` are not added to the process
- * environment. They are sent over the data channel to the
- * already-running worker by the runner (see
- * [[org.apache.spark.api.python.PythonRunner]]).
+ * Note: `pythonIncludes` are not added to the process environment.
+ * Unified execution serializes them in the per-UDF payload delivered
+ * to the already-running worker during Init.
  */
 @Experimental
 object PythonUDFWorkerSpecification {
 
+  private[externalUDF] val ARTIFACTS_RESOURCE_DIRECTORY: String = "artifacts"
+
   /**
    * Creates a [[UDFWorkerSpecification]] from a [[PythonFunction]].
    *
@@ -77,7 +81,7 @@ object PythonUDFWorkerSpecification {
     envVars.put("SPARK_PYTHON_RUNTIME", "PYTHON_WORKER")
     // Enable the execution mode supporting the new UDF execution
     // framework.
-    // TODO [SPARK-55278]: Enable this on the python code
+    // TODO(SPARK-59368): Enable this in the Python worker.
     envVars.put("PYTHON_WORKER_UNIFIED_EXECUTION_ENABLED", "YES")
 
     // Build the ProcessCallable:
@@ -86,8 +90,8 @@ object PythonUDFWorkerSpecification {
     callable.addCommand(func.pythonExec)
     callable.addCommand("-m")
     callable.addCommand(workerModule)
-    // TODO [SPARK-55278]: Add additional, python specific env vars
-    // or transform them into init-message fields
+    // TODO(SPARK-59368): Add Python-specific environment variables or expose them as
+    // Init fields.
     envVars.forEach((k, v) => callable.putEnvironmentVariables(k, v))
 
     // Capabilities: ARROW data format, bidirectional streaming
@@ -107,10 +111,30 @@ object PythonUDFWorkerSpecification {
       .setRunner(callable)
       .setProperties(props)
 
+    val session = WorkerSessionSpecification.newBuilder()
+      .addRequiredResourceDirectories(ARTIFACTS_RESOURCE_DIRECTORY)
+    val pythonSqlConfEntries = ArrowPythonRunner.getPythonRunnerConfEntries
+      .filterNot(_.key == SQLConf.SESSION_LOCAL_TIMEZONE.key)
+      .groupBy(_.key)
+      .values
+      .map(_.head)
+      .toSeq
+      .sortBy(_.key)
+    pythonSqlConfEntries.foreach { entry =>
+      session.putDynamicConfig(
+        entry.key,
+        dynamicConfigRequirement(!entry.isInstanceOf[OptionalConfigEntry[_]]))
+    }
+
     UDFWorkerSpecification.newBuilder()
       .setEnvironment(WorkerEnvironment.newBuilder())
       .setCapabilities(caps)
+      .setSession(session)
       .setDirect(direct)
       .build()
   }
+
+  private def dynamicConfigRequirement(isRequired: Boolean): DynamicConfigRequirement = {
+    DynamicConfigRequirement.newBuilder().setIsRequired(isRequired).build()
+  }
 }
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowPythonRunner.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowPythonRunner.scala
index bbe1b278936a8..b626eaf00326a 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowPythonRunner.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/python/ArrowPythonRunner.scala
@@ -177,25 +177,28 @@ object ArrowPythonRunner {
     }
   }
 
+  private val pythonRunnerConfEntries: Seq[ConfigEntry[_]] = Seq(
+    SQLConf.SESSION_LOCAL_TIMEZONE,
+    SQLConf.PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_NAME,
+    SQLConf.PANDAS_ARROW_SAFE_TYPE_CONVERSION,
+    SQLConf.ARROW_EXECUTION_USE_LARGE_VAR_TYPES,
+    SQLConf.PYTHON_TABLE_UDF_LEGACY_PANDAS_CONVERSION_ENABLED,
+    SQLConf.PYTHON_UDF_LEGACY_PANDAS_CONVERSION_ENABLED,
+    SQLConf.PYTHON_UDF_MAP_IN_BATCH_LEGACY_ACCEPT_ANY_ITERABLE_ENABLED,
+    SQLConf.PYTHON_UDF_PANDAS_INT_TO_DECIMAL_COERCION_ENABLED,
+    SQLConf.PYTHON_UDF_PANDAS_PREFER_INT_EXTENSION_DTYPE,
+    SQLConf.PYSPARK_BINARY_AS_BYTES,
+    // Optional
+    SQLConf.PYTHON_UDF_ARROW_CONCURRENCY_LEVEL,
+    SQLConf.PYTHON_UDF_PROFILER,
+    SQLConf.PYTHON_DATA_SOURCE_PROFILER)
+
+  private[sql] def getPythonRunnerConfEntries: Seq[ConfigEntry[_]] = pythonRunnerConfEntries
+
   /** Return Map with conf settings to be used in ArrowPythonRunner */
   def getPythonRunnerConfMap(conf: SQLConf): Map[String, String] = {
     val confMap = collection.mutable.Map.empty[String, String]
-    Seq(
-      SQLConf.SESSION_LOCAL_TIMEZONE,
-      SQLConf.PANDAS_GROUPED_MAP_ASSIGN_COLUMNS_BY_NAME,
-      SQLConf.PANDAS_ARROW_SAFE_TYPE_CONVERSION,
-      SQLConf.ARROW_EXECUTION_USE_LARGE_VAR_TYPES,
-      SQLConf.PYTHON_TABLE_UDF_LEGACY_PANDAS_CONVERSION_ENABLED,
-      SQLConf.PYTHON_UDF_LEGACY_PANDAS_CONVERSION_ENABLED,
-      SQLConf.PYTHON_UDF_MAP_IN_BATCH_LEGACY_ACCEPT_ANY_ITERABLE_ENABLED,
-      SQLConf.PYTHON_UDF_PANDAS_INT_TO_DECIMAL_COERCION_ENABLED,
-      SQLConf.PYTHON_UDF_PANDAS_PREFER_INT_EXTENSION_DTYPE,
-      SQLConf.PYSPARK_BINARY_AS_BYTES,
-      // Optional
-      SQLConf.PYTHON_UDF_ARROW_CONCURRENCY_LEVEL,
-      SQLConf.PYTHON_UDF_PROFILER,
-      SQLConf.PYTHON_DATA_SOURCE_PROFILER
-    ).foreach {
+    pythonRunnerConfEntries.foreach {
       case c: OptionalConfigEntry[_] =>
         conf.getConf(c).foreach(v => confMap.update(c.key, v.toString))
       case c: ConfigEntry[_] =>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala
new file mode 100644
index 0000000000000..976fa56fd9b83
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala
@@ -0,0 +1,450 @@
+/*
+ * 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.externalUDF
+
+import java.nio.ByteBuffer
+
+import scala.jdk.CollectionConverters._
+
+import org.apache.spark.api.python.{PythonEvalType, SimplePythonFunction}
+import org.apache.spark.sql.QueryTest
+import org.apache.spark.sql.catalyst.expressions.{Expression, ExternalUDFInitContext,
+  ExternalUserDefinedFunction, Literal, NamedArgumentExpression, NamedExpression, PythonUDF}
+import org.apache.spark.sql.execution.python.ArrowPythonRunner
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DataType, IntegerType, LongType, StringType, StructField,
+  StructType}
+import org.apache.spark.udf.worker.{DynamicConfigRequirement, Init, UDFWorkerDataFormat,
+  UDFWorkerSpecification, WorkerContextReference, WorkerSessionSpecification}
+
+class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkSession {
+
+  private def contextReference(
+      source: String,
+      defaultValue: Option[String] = None): WorkerContextReference = {
+    val builder = WorkerContextReference.newBuilder().setSource(source)
+    defaultValue.foreach(builder.setDefaultValue)
+    builder.build()
+  }
+
+  private def dynamicConfigRequirement(isRequired: Boolean): DynamicConfigRequirement = {
+    DynamicConfigRequirement.newBuilder().setIsRequired(isRequired).build()
+  }
+
+  private def expectExternalUDF(expr: Expression): ExternalUserDefinedFunction = expr match {
+    case udf: ExternalUserDefinedFunction => udf
+    case other => fail(s"Expected ExternalUserDefinedFunction, found ${other.getClass.getName}")
+  }
+
+  private def initContext(
+      taskContext: Map[String, String] = Map(
+        "partitionId" -> "3",
+        ExternalUDFInitContext.DRIVER_ID_CONTEXT_KEY -> "driver",
+        ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY -> "true"),
+      environmentVariables: Map[String, String] = Map.empty,
+      dynamicConfig: Map[String, String] = Map.empty,
+      resourceDirectories: Map[String, String] = Map(
+        PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY ->
+          "/var/resources/artifact-1"),
+      timezone: String = "America/Los_Angeles"): ExternalUDFInitContext = {
+    ExternalUDFInitContext(
+      protocolVersion = 1,
+      dataFormat = UDFWorkerDataFormat.ARROW,
+      inputSchema = Array[Byte](1, 2),
+      outputSchema = Array[Byte](3, 4),
+      timezone = timezone,
+      taskContext = taskContext,
+      environmentVariables = environmentVariables,
+      dynamicConfig = dynamicConfig,
+      resourceDirectories = resourceDirectories)
+  }
+
+  private def pythonFunction(
+      command: Array[Byte] = Array[Byte](10, 20, 30),
+      pythonIncludes: Seq[String] = Seq("first.zip", "second.zip"),
+      pythonVersion: String = "3.12"): SimplePythonFunction = {
+    new SimplePythonFunction(
+      command = command,
+      envVars = Map.empty[String, String].asJava,
+      pythonIncludes = pythonIncludes.asJava,
+      pythonExec = "python3",
+      pythonVer = pythonVersion,
+      broadcastVars = null,
+      accumulator = null)
+  }
+
+  private def pythonUDF(
+      children: Seq[Expression],
+      evalType: Int = PythonEvalType.SQL_ARROW_BATCHED_UDF): PythonUDF = {
+    PythonUDF(
+      name = "plus_one",
+      func = pythonFunction(),
+      dataType = IntegerType,
+      children = children,
+      evalType = evalType,
+      udfDeterministic = true)
+  }
+
+  private def genericFunction(
+      session: WorkerSessionSpecification.Builder): ExternalUserDefinedFunction = {
+    ExternalUserDefinedFunction(
+      name = Some("generic"),
+      workerSpec = UDFWorkerSpecification.newBuilder().setSession(session).build(),
+      payload = Array[Byte](5, 6),
+      dataType = IntegerType,
+      children = Seq(Literal(1)),
+      udfDeterministic = true,
+      udfNullable = false)
+  }
+
+  private def pythonDynamicConfig: Map[String, String] = {
+    ArrowPythonRunner.getPythonRunnerConfMap(spark.sessionState.conf) -
+      SQLConf.SESSION_LOCAL_TIMEZONE.key
+  }
+
+  private def hexBytes(value: String): Array[Byte] = {
+    value.grouped(2).map(pair => Integer.parseInt(pair, 16).toByte).toArray
+  }
+
+  test("generic Init resolves only worker-declared context") {
+    val session = WorkerSessionSpecification.newBuilder()
+      .putStaticConfig("static", "static-value")
+      .putEnvironmentVariableReferences("forwarded.env", contextReference("SOURCE_ENV"))
+      .putEnvironmentVariableReferences(
+        "defaulted.env",
+        contextReference("MISSING_ENV", Some("default-value")))
+      .putEnvironmentVariableReferences("optional.env", contextReference("MISSING_OPTIONAL_ENV"))
+      .putDynamicConfig("forwarded.dynamic", dynamicConfigRequirement(isRequired = true))
+      .putDynamicConfig("optional.dynamic", dynamicConfigRequirement(isRequired = false))
+      .addRequiredResourceDirectories("inputs")
+    val function = genericFunction(session)
+
+    val init = function.buildInit(initContext(
+      environmentVariables = Map("SOURCE_ENV" -> "env-value", "SECRET" -> "not-forwarded"),
+      dynamicConfig = Map(
+        "forwarded.dynamic" -> "dynamic-value",
+        "unrequested.dynamic" -> "not-forwarded"),
+      resourceDirectories = Map(
+        "inputs" -> "/var/resources/input-data",
+        "unrequested" -> "/var/resources/private")))
+
+    assert(init.getProtocolVersion === 1)
+    assert(init.getDataFormat === UDFWorkerDataFormat.ARROW)
+    assert(init.getInputSchema.toByteArray.sameElements(Array[Byte](1, 2)))
+    assert(init.getOutputSchema.toByteArray.sameElements(Array[Byte](3, 4)))
+    assert(init.getTimezone === "America/Los_Angeles")
+    assert(init.getTaskContextMap.asScala.toMap === Map(
+      "partitionId" -> "3",
+      ExternalUDFInitContext.DRIVER_ID_CONTEXT_KEY -> "driver",
+      ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY -> "true"))
+    assert(init.getEnvironmentVariablesMap.asScala.toMap ===
+      Map("forwarded.env" -> "env-value", "defaulted.env" -> "default-value"))
+    assert(init.getSessionConfMap.asScala.toMap === Map(
+      "static" -> "static-value",
+      "forwarded.dynamic" -> "dynamic-value"))
+    assert(init.getResourceDirectoriesMap.asScala.toMap ===
+      Map("inputs" -> "/var/resources/input-data"))
+    assert(init.getUdf.getName === "generic")
+    assert(init.getUdf.getFormat === "raw-v1")
+    assert(init.getUdf.getInvocationId === function.resultId.id)
+    assert(init.getUdf.getPayload.toByteArray.sameElements(Array[Byte](5, 6)))
+    assert(init.getUdf.getInput.getSchemaFormat === "spark-sql-data-type-json-v1")
+    assert(init.getUdf.getInput.getArguments(0).getInputOffset === 0)
+    assert(!init.getUdf.getInput.getArguments(0).hasName)
+    assert(!init.getUdf.hasEvalType)
+    assert(!init.hasParameters)
+  }
+
+  test("generic Init rejects conflicting session context targets") {
+    val session = WorkerSessionSpecification.newBuilder()
+      .putStaticConfig("duplicate", "static")
+      .putDynamicConfig("duplicate", dynamicConfigRequirement(isRequired = false))
+    val function = genericFunction(session)
+
+    val error = intercept[IllegalArgumentException] {
+      function.buildInit(initContext())
+    }
+    assert(error.getMessage.contains("duplicate target keys: duplicate"))
+  }
+
+  test("generic Init rejects empty worker context references") {
+    val session = WorkerSessionSpecification.newBuilder()
+      .putEnvironmentVariableReferences("target", contextReference(""))
+    val function = genericFunction(session)
+
+    val error = intercept[IllegalArgumentException] {
+      function.buildInit(initContext())
+    }
+    assert(error.getMessage.contains("Empty worker context source for target"))
+
+    val emptyConfigName = WorkerSessionSpecification.newBuilder()
+      .putDynamicConfig("", dynamicConfigRequirement(isRequired = false))
+    val emptyNameError = intercept[IllegalArgumentException] {
+      genericFunction(emptyConfigName).buildInit(initContext())
+    }
+    assert(emptyNameError.getMessage.contains("empty target key"))
+  }
+
+  test("generic Init enforces dynamic configuration requirements") {
+    val session = WorkerSessionSpecification.newBuilder()
+      .putDynamicConfig("required.b", dynamicConfigRequirement(isRequired = true))
+      .putDynamicConfig("optional", dynamicConfigRequirement(isRequired = false))
+      .putDynamicConfig("required.a", dynamicConfigRequirement(isRequired = true))
+    val function = genericFunction(session)
+
+    val error = intercept[IllegalArgumentException] {
+      function.buildInit(initContext(dynamicConfig = Map("optional" -> "present")))
+    }
+    assert(error.getMessage.contains(
+      "Missing required dynamic configuration: required.a, required.b"))
+
+    val init = function.buildInit(initContext(dynamicConfig = Map(
+      "required.a" -> "",
+      "required.b" -> "value",
+      "optional" -> "optional-value",
+      "unrequested" -> "not-forwarded")))
+    assert(init.getSessionConfMap.asScala.toMap === Map(
+      "required.a" -> "",
+      "required.b" -> "value",
+      "optional" -> "optional-value"))
+  }
+
+  test("generic Init validates required resource directories") {
+    def build(requiredNames: String*)(resources: Map[String, String]): Init = {
+      val session = WorkerSessionSpecification.newBuilder()
+        .addAllRequiredResourceDirectories(requiredNames.asJava)
+      genericFunction(session).buildInit(initContext(resourceDirectories = resources))
+    }
+
+    val emptyName = intercept[IllegalArgumentException] {
+      build("")(Map.empty)
+    }
+    assert(emptyName.getMessage.contains("empty required resource directory name"))
+
+    val duplicateNames = intercept[IllegalArgumentException] {
+      build("input", "input")(Map("input" -> "/var/resources/input"))
+    }
+    assert(duplicateNames.getMessage.contains(
+      "duplicate required resource directories: input"))
+
+    val missing = intercept[IllegalArgumentException] {
+      build("input")(Map.empty)
+    }
+    assert(missing.getMessage.contains("Missing required resource directories: input"))
+
+    val emptyPath = intercept[IllegalArgumentException] {
+      build("input")(Map("input" -> ""))
+    }
+    assert(emptyPath.getMessage.contains("Empty required resource directories: input"))
+
+    assert(build()(Map("unrequested" -> "/var/resources/private"))
+      .getResourceDirectoriesMap.isEmpty)
+    val resolved = build("input", "cache")(Map(
+      "input" -> "/var/resources/input",
+      "cache" -> "/var/resources/cache",
+      "unrequested" -> "/var/resources/private"))
+    assert(resolved.getResourceDirectoriesMap.asScala.toMap === Map(
+      "input" -> "/var/resources/input",
+      "cache" -> "/var/resources/cache"))
+  }
+
+  test("PySpark conversion uses generic Init and refreshes rewritten input metadata") {
+    val udf = pythonUDF(Seq(
+      Literal(1),
+      NamedArgumentExpression("named", Literal("value"))))
+    val workerSpec = PythonUDFWorkerSpecification.fromPythonFunction(
+      udf.func,
+      spark.sparkContext.getConf)
+    val external = PythonExternalUserDefinedFunction.fromPythonUDF(
+      udf,
+      workerSpec)
+    val pythonTaskContext = Map(
+      "partitionId" -> "3",
+      ExternalUDFInitContext.DRIVER_ID_CONTEXT_KEY -> "driver",
+      ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY -> "false")
+    val currentDynamicConfig =
+      pythonDynamicConfig.updated(SQLConf.PYSPARK_BINARY_AS_BYTES.key, "false")
+    val init = external.buildInit(initContext(
+      taskContext = pythonTaskContext,
+      environmentVariables = Map("UNREQUESTED_SECRET" -> "not-forwarded"),
+      dynamicConfig = currentDynamicConfig))
+
+    assert(external.getClass === classOf[ExternalUserDefinedFunction])
+    assert(init.getUdf.getName === "plus_one")
+    assert(init.getUdf.getFormat === "pyspark-udf-v2")
+    assert(init.getUdf.getEvalType === PythonEvalType.SQL_ARROW_BATCHED_UDF.toString)
+    assert(init.getUdf.getInvocationId === udf.resultId.id)
+    assert(init.getEnvironmentVariablesMap.isEmpty)
+    assert(init.getTaskContextMap.get(
+      ExternalUDFInitContext.DRIVER_ID_CONTEXT_KEY) === "driver")
+    assert(init.getTaskContextMap.get(
+      ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY) === "false")
+    assert(init.getSessionConfMap.get(SQLConf.PYSPARK_BINARY_AS_BYTES.key) === "false")
+    assert((pythonDynamicConfig - SQLConf.PYSPARK_BINARY_AS_BYTES.key).forall { case (key, value) =>
+      init.getSessionConfMap.get(key) == value
+    })
+    assert(!init.getSessionConfMap.containsKey(SQLConf.SESSION_LOCAL_TIMEZONE.key))
+    assert(!init.hasParameters)
+
+    val nextTaskContext = pythonTaskContext
+      .updated("partitionId", "4")
+      .updated(ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY, "true")
+    val nextInit = external.buildInit(initContext(
+      taskContext = nextTaskContext,
+      dynamicConfig = pythonDynamicConfig,
+      resourceDirectories = Map(
+        PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY ->
+          "/var/resources/artifact-2"),
+      timezone = "UTC"))
+    assert(nextInit.getUdf === init.getUdf)
+    assert(nextInit.getTaskContextMap.get("partitionId") === "4")
+    assert(nextInit.getTaskContextMap.get(
+      ExternalUDFInitContext.DRIVER_ID_CONTEXT_KEY) === "driver")
+    assert(nextInit.getTaskContextMap.get(
+      ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY) === "true")
+    assert(nextInit.getResourceDirectoriesMap.get(
+      PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY) ===
+      "/var/resources/artifact-2")
+    assert(nextInit.getTimezone === "UTC")
+
+    val decoded = PythonUDFPayload.decode(external.payload)
+    assert(decoded.command.sameElements(Array[Byte](10, 20, 30)))
+    assert(decoded.pythonIncludes === Vector("first.zip", "second.zip"))
+    assert(decoded.pythonVersion === "3.12")
+
+    val inputSchema = DataType.fromJson(init.getUdf.getInput.getSchema.toStringUtf8)
+    assert(inputSchema === StructType(Seq(
+      StructField("_0", IntegerType, nullable = false),
+      StructField("_1", StringType, nullable = false))))
+    assert(init.getUdf.getInput.getArgumentsList.asScala.map(_.getInputOffset) === Seq(0, 1))
+    assert(!init.getUdf.getInput.getArguments(0).hasName)
+    assert(init.getUdf.getInput.getArguments(1).getName === "named")
+
+    val rewritten = expectExternalUDF(external.withNewChildren(Seq(
+      Literal(1L),
+      NamedArgumentExpression("renamed", Literal("value")))))
+    val rewrittenInit = rewritten.buildInit(initContext(dynamicConfig = pythonDynamicConfig))
+    val rewrittenSchema =
+      DataType.fromJson(rewrittenInit.getUdf.getInput.getSchema.toStringUtf8)
+    assert(rewritten.payload eq external.payload)
+    assert(rewrittenInit.getUdf.getInvocationId === init.getUdf.getInvocationId)
+    assert(rewrittenInit.getUdf.getInput.getSchemaFormat === "spark-sql-data-type-json-v1")
+    assert(rewrittenSchema === StructType(Seq(
+      StructField("_0", LongType, nullable = false),
+      StructField("_1", StringType, nullable = false))))
+    val rewrittenArguments = rewrittenInit.getUdf.getInput.getArgumentsList.asScala.map { arg =>
+      (arg.getInputOffset, if (arg.hasName) Some(arg.getName) else None)
+    }
+    assert(rewrittenArguments === Seq((0, None), (1, Some("renamed"))))
+
+    val canonicalized = expectExternalUDF(external.canonicalized)
+    assert(canonicalized.resultId.id === -1L)
+    assert(canonicalized.payload eq external.payload)
+    assert(external.semanticEquals(external.copy(resultId = NamedExpression.newExprId)))
+  }
+
+  test("Python worker specification declares its generic session requirements") {
+    val spec = PythonUDFWorkerSpecification.fromPythonFunction(
+      pythonFunction(),
+      spark.sparkContext.getConf)
+    val session = spec.getSession
+    val dynamicConfig = session.getDynamicConfigMap.asScala.toMap
+
+    val expectedDynamicKeys = ArrowPythonRunner.getPythonRunnerConfEntries
+      .map(_.key)
+      .filterNot(_ == SQLConf.SESSION_LOCAL_TIMEZONE.key)
+      .toSet
+    assert(dynamicConfig.keySet === expectedDynamicKeys)
+    val optionalKeys = Set(
+      SQLConf.PYTHON_UDF_ARROW_CONCURRENCY_LEVEL.key,
+      SQLConf.PYTHON_UDF_PROFILER.key,
+      SQLConf.PYTHON_DATA_SOURCE_PROFILER.key)
+    assert(dynamicConfig.forall { case (name, requirement) =>
+      requirement.getIsRequired === !optionalKeys.contains(name)
+    })
+    assert(session.getRequiredResourceDirectoriesList.asScala.toSeq ===
+      Seq(PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY))
+    assert(session.getStaticConfigMap.isEmpty)
+    val defaultFunction = PythonExternalUserDefinedFunction.fromPythonUDF(
+      pythonUDF(Seq(Literal(1))),
+      spec)
+    val defaultInit = defaultFunction.buildInit(initContext(
+      dynamicConfig = pythonDynamicConfig,
+      resourceDirectories = Map(
+        PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY ->
+          "/var/resources/artifact-default")))
+    assert(defaultInit.getSessionConfMap.get(SQLConf.PYSPARK_BINARY_AS_BYTES.key) === "true")
+    assert(!defaultInit.getSessionConfMap.containsKey(
+      SQLConf.PYTHON_UDF_ARROW_CONCURRENCY_LEVEL.key))
+    assert(defaultInit.getResourceDirectoriesMap.get(
+      PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY) ===
+      "/var/resources/artifact-default")
+
+    val missingConfig = intercept[IllegalArgumentException] {
+      defaultFunction.buildInit(initContext())
+    }
+    assert(missingConfig.getMessage.contains("Missing required dynamic configuration"))
+
+    Seq(
+      Map.empty[String, String],
+      Map(PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY -> "")
+    ).foreach { resourceDirectories =>
+      val error = intercept[IllegalArgumentException] {
+        defaultFunction.buildInit(initContext(
+          dynamicConfig = pythonDynamicConfig,
+          resourceDirectories = resourceDirectories))
+      }
+      assert(error.getMessage.contains("resource director"))
+    }
+    assert(session.getEnvironmentVariableReferencesMap.isEmpty)
+  }
+
+  test("PySpark conversion rejects unsupported evaluation types") {
+    val udf = pythonUDF(Seq(Literal(1)), PythonEvalType.SQL_BATCHED_UDF)
+    val error = intercept[IllegalArgumentException] {
+      PythonExternalUserDefinedFunction.fromPythonUDF(
+        udf,
+        UDFWorkerSpecification.getDefaultInstance)
+    }
+    assert(error.getMessage.contains("Unsupported Python external UDF eval type"))
+  }
+
+  test("Python payload encoding has a stable versioned wire format") {
+    val payload = PythonUDFPayload.encode(pythonFunction(
+      command = Array[Byte](1),
+      pythonIncludes = Seq("x"),
+      pythonVersion = "v"))
+    val expectedPayload = hexBytes(
+      "505955440000000200000001010000000100000001780000000176")
+    assert(payload.sameElements(expectedPayload))
+
+    Seq(1, 3).foreach { version =>
+      val unsupportedVersion = payload.clone()
+      ByteBuffer.wrap(unsupportedVersion).putInt(Integer.BYTES, version)
+      assert(intercept[IllegalArgumentException] {
+        PythonUDFPayload.decode(unsupportedVersion)
+      }.getMessage.contains(s"unsupported version $version"))
+    }
+    assert(intercept[IllegalArgumentException] {
+      PythonUDFPayload.decode(payload :+ 0.toByte)
+    }.getMessage.contains("trailing bytes"))
+    assert(intercept[IllegalArgumentException] {
+      PythonUDFPayload.decode(payload.dropRight(1))
+    }.getMessage.contains("invalid Python version length"))
+  }
+}
diff --git a/udf/worker/proto/src/main/protobuf/udf_message.proto b/udf/worker/proto/src/main/protobuf/udf_message.proto
index dd7c9ceb963b6..297805af81deb 100644
--- a/udf/worker/proto/src/main/protobuf/udf_message.proto
+++ b/udf/worker/proto/src/main/protobuf/udf_message.proto
@@ -152,13 +152,11 @@ message UdfControlResponse {
 //     [[input_schema]] / [[output_schema]] -- matching the worker
 //     spec, not the function's view) and what per-session
 //     context the worker needs ([[timezone]], [[session_conf]],
-//     [[task_context]], [[parameters]]).
-//   * [[UdfPayload]] carries everything the client side of Spark
-//     (where the UDF is defined and serialized) packs -- the
-//     serialized callable, an opaque format tag, and any encoder
-//     metadata bundled with the callable. The wire protocol does
-//     not enumerate encoder shapes; that is left to the client and
-//     worker to agree on per UDF type.
+//     [[task_context]], [[environment_variables]],
+//     [[resource_directories]], and [[parameters]]).
+//   * [[UdfPayload]] carries the serialized callable plus metadata
+//     for this invocation. The callable stays opaque; common
+//     invocation metadata can use typed fields on [[UdfPayload]].
 message Init {
     // (Optional) Protocol version declared by the engine for this stream.
     // Allows the worker to detect version mismatches early and reject
@@ -197,10 +195,10 @@ message Init {
     // describes the bytes the engine will actually put on
     // [[DataRequest.data]] for this session, matching what the
     // worker advertised in its spec. It is NOT necessarily the
-    // schema the function definer expressed; the UDF's own type
-    // information lives inside [[UdfPayload]], typically embedded
-    // alongside the callable in [[UdfPayload.payload]] (e.g. as
-    // input/output encoders chosen per UDF type).
+    // schema the function definer expressed. Logical input types and
+    // argument bindings live in [[UdfPayload.input]]. Language-private
+    // metadata, such as callable serialization and UDF-specific
+    // encoders, may remain in [[UdfPayload.payload]].
     //
     // Left unset when the worker can derive the schema from the
     // payload alone.
@@ -216,13 +214,17 @@ message Init {
     // provided by the engine. Common keys identify the task instance
     // for diagnostics, logging, and stateful workers -- e.g.
     // partition id, task attempt id, stage id, micro-batch id.
+    // Spark physical execution must also include "driverId" (defaulting
+    // to "driver") and "isDriver" (derived from the local SparkEnv) for
+    // every external worker session.
     // Engine and worker agree on the keys they share; the protocol
     // does not enumerate them.
     map<string, string> task_context = 7;
 
-    // (Optional; defaults to an empty map.) Worker-private knobs not
-    // already captured by typed fields above. Free-form; both sides
-    // agree on the keys they need.
+    // (Optional; defaults to an empty map.) Worker-private configuration
+    // not already captured by typed fields above. This combines literal
+    // WorkerSessionSpecification.static_config values with the dynamic
+    // values selected by WorkerSessionSpecification.dynamic_config.
     //
     // Any key that two languages converge on is a candidate for
     // promotion to a structured proto field -- once promoted, it gets
@@ -242,13 +244,26 @@ message Init {
     // Spark does.
     optional string timezone = 9;
 
+    // (Optional; defaults to an empty map.) Environment variables
+    // requested by the worker specification for this session. The
+    // engine only forwards explicitly requested names; it MUST NOT
+    // expose its complete process environment.
+    map<string, string> environment_variables = 10;
+
+    // (Optional; defaults to an empty map.) Logical resource name -> fully
+    // resolved local directory in the worker's filesystem view. The engine
+    // MUST include every non-empty name declared in
+    // WorkerSessionSpecification.required_resource_directories with a
+    // non-empty path, and MUST NOT include unrequested resources.
+    map<string, string> resource_directories = 11;
+
     // Reserved for future typed Init fields, in particular keys
     // graduated from [[session_conf]] (see the [[timezone]] precedent
     // above). Numbers >= 100 are intentionally NOT reserved here; if
     // a future revision needs an opaque escape-hatch field, give it a
     // number >= 100 alongside [[parameters]] and add a field-level
     // comment so the convention stays visible.
-    reserved 10 to 99;
+    reserved 12 to 99;
 
     // (Optional) Engine-packed opaque parameters specific to a
     // particular kind of UDF execution. The escape hatch for
@@ -261,11 +276,11 @@ message Init {
     //
     // Numbers >= 100 are reserved by convention for opaque
     // escape-hatch fields like this one; new typed fields use the
-    // reserved 10..99 range.
+    // reserved 12..99 range.
     //
     // Client-side init data (anything packed by the layer that
     // defines and serializes the UDF) does NOT belong here -- it
-    // travels inside [[UdfPayload.payload]] instead.
+    // travels inside [[UdfPayload]] instead.
     optional bytes parameters = 100;
 }
 
@@ -535,10 +550,10 @@ message ProtocolError {
     string message = 1;
 }
 
-// The single UDF body delivered to the worker on [[Init]]. Opaque to
-// the engine: the engine forwards [[payload]] and [[format]]
-// unchanged, and the worker decodes them per the format the client
-// and worker have agreed on.
+// The single UDF body delivered to the worker on [[Init]]. The
+// serialized [[payload]] is opaque to the engine and decoded by the
+// worker according to [[format]]. Other fields describe common
+// invocation metadata without making the engine language-aware.
 message UdfPayload {
     // (Required, may be empty when chunked.) Serialized UDF bundle,
     // opaque to the engine. The encoding is declared in [[format]].
@@ -590,6 +605,39 @@ message UdfPayload {
     // left unset. Otherwise the client side of the protocol sets it
     // explicitly.
     optional string eval_type = 5;
+
+    // (Optional) Engine-assigned identity of this UDF invocation.
+    // This is metadata about the invocation rather than part of the
+    // opaque language-specific payload.
+    optional int64 invocation_id = 7;
+
+    // (Optional) Logical input metadata for this invocation. This is
+    // distinct from Init.input_schema, which describes the encoded wire
+    // data. The engine rebuilds this metadata when expression children
+    // are rewritten.
+    optional UdfInputMetadata input = 8;
+}
+
+// Language-independent description of a UDF invocation's logical inputs.
+message UdfInputMetadata {
+    // (Required, non-empty.) Format tag for [[schema]]. The client and
+    // worker agree on the format; Spark SQL uses
+    // "spark-sql-data-type-json-v1".
+    string schema_format = 1;
+
+    // Logical input schema encoded in [[schema_format]].
+    bytes schema = 2;
+
+    // Positional and named bindings into [[schema]].
+    repeated UdfArgument arguments = 3;
+}
+
+message UdfArgument {
+    // Zero-based field offset in [[UdfInputMetadata.schema]].
+    uint32 input_offset = 1;
+
+    // Optional named-argument name supplied by the caller.
+    optional string name = 2;
 }
 
 // =====================================================================
diff --git a/udf/worker/proto/src/main/protobuf/worker_spec.proto b/udf/worker/proto/src/main/protobuf/worker_spec.proto
index 83dac4f962e5f..70402bc32891a 100644
--- a/udf/worker/proto/src/main/protobuf/worker_spec.proto
+++ b/udf/worker/proto/src/main/protobuf/worker_spec.proto
@@ -33,6 +33,14 @@ message UDFWorkerSpecification {
     // (Required)
     WorkerCapabilities capabilities = 2;
 
+    // Describes the context to include in every Init message sent to
+    // this worker. The declaration is part of the worker specification
+    // so the engine can construct Init without knowing the worker's
+    // implementation language.
+    //
+    // (Optional)
+    WorkerSessionSpecification session = 4;
+
     // How to create new workers.
     // At the moment, only direct creation is supported.
     // This can be extended with indirect/provisioned creation in the future.
@@ -43,6 +51,58 @@ message UDFWorkerSpecification {
     }
 }
 
+// Static values and engine context required to initialize one execution
+// session on a worker. The engine resolves supported, authorized environment,
+// configuration, and resource names when it constructs Init. A declaration
+// selects a value but does not grant access to it.
+message WorkerSessionSpecification {
+    // Static worker-private settings copied to Init.session_conf.
+    //
+    // (Optional)
+    map<string, string> static_config = 1;
+
+    // Non-empty Init output key -> literal engine process environment
+    // variable reference. The engine must authorize the non-empty source
+    // before making its value available; this worker declaration is
+    // selection, not authorization. Derived engine or task state is not an
+    // environment-variable source.
+    //
+    // (Optional)
+    map<string, WorkerContextReference> environment_variable_references = 2;
+
+    // Non-empty logical name -> dynamic configuration requirement. The engine
+    // looks up the same name in its authorized dynamic configuration and, when
+    // available, copies the value to Init.session_conf under that name.
+    //
+    // (Optional)
+    map<string, DynamicConfigRequirement> dynamic_config = 3;
+
+    // Non-empty, unique logical names of local resource directories required
+    // in Init.resource_directories. The engine resolves each name to a fully
+    // qualified path in the worker's filesystem view. A declaration selects an
+    // engine-authorized resource; it does not grant filesystem access.
+    //
+    // (Optional)
+    repeated string required_resource_directories = 4;
+}
+
+message DynamicConfigRequirement {
+    // Whether session initialization must fail when the engine cannot provide
+    // this named value. Missing optional values are omitted from Init.
+    //
+    // (Optional; defaults to false.)
+    bool is_required = 1;
+}
+
+message WorkerContextReference {
+    // (Required, non-empty.) Source environment variable name.
+    string source = 1;
+
+    // Value to use when the source is unavailable. When unset, a missing
+    // source is omitted from Init.
+    optional string default_value = 2;
+}
+
 // Set of callables that can be used to setup, verify,
 // and cleanup the worker environment before the worker
 // callable is invoked.

From 7c78517581cec3bd40354264faf5ecfb9e9cba56 Mon Sep 17 00:00:00 2001
From: Haiyang Sun <75672016+haiyangsun-db@users.noreply.github.com>
Date: Thu, 10 Sep 2026 20:22:42 +0000
Subject: [PATCH 2/2] [SPARK-59364][UDF] Clarify experimental Python adapter
 naming

---
 .../externalUDF/ExternalUDFPlanner.scala      |  2 +-
 ...n.scala => PythonExternalUDFAdapter.scala} | 13 ++++---
 .../externalUDF/PythonUDFPayload.scala        |  4 +--
 ...scala => PythonUDFWorkerSpecBuilder.scala} | 12 +++----
 ...a => ExternalUDFInitializationSuite.scala} | 34 +++++++++----------
 ... => PythonUDFWorkerSpecBuilderSuite.scala} |  8 ++---
 6 files changed, 38 insertions(+), 35 deletions(-)
 rename sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/{PythonExternalUserDefinedFunction.scala => PythonExternalUDFAdapter.scala} (84%)
 rename sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/{PythonUDFWorkerSpecification.scala => PythonUDFWorkerSpecBuilder.scala} (94%)
 rename sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/{PythonExternalUserDefinedFunctionSuite.scala => ExternalUDFInitializationSuite.scala} (94%)
 rename sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/{PythonUDFWorkerSpecificationSuite.scala => PythonUDFWorkerSpecBuilderSuite.scala} (96%)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFPlanner.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFPlanner.scala
index 06e119d307b97..3017ffc616e1d 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFPlanner.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFPlanner.scala
@@ -132,7 +132,7 @@ class UnifiedExternalUDFPlanner(
       profile: Option[ResourceProfile]): LogicalPlan = {
     val pythonUdf = func.asInstanceOf[PythonUDF]
     val workerSpec =
-      PythonUDFWorkerSpecification.fromPythonFunction(
+      PythonUDFWorkerSpecBuilder.build(
         pythonUdf.func, conf)
     val udf = ExternalUserDefinedFunction(
       name = Some(pythonUdf.name),
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUDFAdapter.scala
similarity index 84%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala
rename to sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUDFAdapter.scala
index f9791ffbabcb5..b69f4daaf14b6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunction.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUDFAdapter.scala
@@ -20,15 +20,20 @@ import org.apache.spark.api.python.PythonEvalType
 import org.apache.spark.sql.catalyst.expressions.{ExternalUserDefinedFunction, PythonUDF}
 import org.apache.spark.udf.worker.UDFWorkerSpecification
 
-/** Converts PySpark UDF metadata to the language-neutral external UDF representation. */
-private[sql] object PythonExternalUserDefinedFunction {
-  val PAYLOAD_FORMAT: String = "pyspark-udf-v2"
+/**
+ * Adapts PySpark UDF metadata to the language-neutral external UDF representation.
+ *
+ * This helper only constructs protocol-facing metadata. Execution remains in the generic
+ * external UDF framework.
+ */
+private[externalUDF] object PythonExternalUDFAdapter {
+  private val PAYLOAD_FORMAT: String = "pyspark-udf-experimental"
 
   /**
    * Creates a scalar Python external UDF while leaving worker launch policy to the caller. Python
    * metadata stays inside the opaque payload; Init construction remains language-independent.
    */
-  def fromPythonUDF(
+  def toExternalUDF(
       udf: PythonUDF,
       workerSpec: UDFWorkerSpecification): ExternalUserDefinedFunction = {
     require(
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala
index 6c70de203ddea..fc424a8e52ae1 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFPayload.scala
@@ -25,7 +25,7 @@ import scala.jdk.CollectionConverters._
 import org.apache.spark.api.python.PythonFunction
 
 /**
- * Versioned static payload for a scalar PySpark UDF.
+ * Experimental, versioned static payload for a scalar PySpark UDF.
  *
  * Only Python-private state which does not change when Catalyst rewrites an invocation is stored
  * here. Logical input types, argument bindings, resource locations, and runtime state are supplied
@@ -42,7 +42,7 @@ private[sql] final class PythonUDFPayload private (
 private[sql] object PythonUDFPayload {
   // ASCII "PYUD" followed by a format version.
   private val MAGIC = 0x50595544
-  private val VERSION = 2
+  private val VERSION = 1
 
   /** Encodes all Python-specific state which is invariant across task attempts. */
   def encode(func: PythonFunction): Array[Byte] = {
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecBuilder.scala
similarity index 94%
rename from sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala
rename to sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecBuilder.scala
index 8a39d11c01fa4..b55aff6fd24a4 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecification.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecBuilder.scala
@@ -19,7 +19,6 @@ package org.apache.spark.sql.execution.externalUDF
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.SparkConf
-import org.apache.spark.annotation.Experimental
 import org.apache.spark.api.python.{PythonFunction, PythonUtils}
 import org.apache.spark.internal.config.OptionalConfigEntry
 import org.apache.spark.internal.config.Python.PYTHON_WORKER_MODULE
@@ -28,9 +27,9 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.udf.worker._
 
 /**
- * :: Experimental ::
- * Builds a [[UDFWorkerSpecification]] for Python UDFs from a
- * [[PythonFunction]] and [[SparkConf]].
+ * Builds a [[UDFWorkerSpecification]] for Python UDFs from a [[PythonFunction]] and
+ * [[SparkConf]]. This helper adapts Python launch metadata without adding Python-specific
+ * behavior to the language-neutral worker protocol.
  *
  * Reuses the same information the existing
  * [[org.apache.spark.api.python.PythonWorkerFactory]] uses:
@@ -44,8 +43,7 @@ import org.apache.spark.udf.worker._
  * Unified execution serializes them in the per-UDF payload delivered
  * to the already-running worker during Init.
  */
-@Experimental
-object PythonUDFWorkerSpecification {
+private[externalUDF] object PythonUDFWorkerSpecBuilder {
 
   private[externalUDF] val ARTIFACTS_RESOURCE_DIRECTORY: String = "artifacts"
 
@@ -57,7 +55,7 @@ object PythonUDFWorkerSpecification {
    * @param conf the SparkConf for reading the worker module config
    * @return a fully populated [[UDFWorkerSpecification]]
    */
-  def fromPythonFunction(
+  def build(
       func: PythonFunction,
       conf: SparkConf): UDFWorkerSpecification = {
 
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFInitializationSuite.scala
similarity index 94%
rename from sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala
rename to sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFInitializationSuite.scala
index 976fa56fd9b83..1eeeaddf5e975 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonExternalUserDefinedFunctionSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFInitializationSuite.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.types.{DataType, IntegerType, LongType, StringType,
 import org.apache.spark.udf.worker.{DynamicConfigRequirement, Init, UDFWorkerDataFormat,
   UDFWorkerSpecification, WorkerContextReference, WorkerSessionSpecification}
 
-class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkSession {
+class ExternalUDFInitializationSuite extends QueryTest with SharedSparkSession {
 
   private def contextReference(
       source: String,
@@ -59,7 +59,7 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
       environmentVariables: Map[String, String] = Map.empty,
       dynamicConfig: Map[String, String] = Map.empty,
       resourceDirectories: Map[String, String] = Map(
-        PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY ->
+        PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY ->
           "/var/resources/artifact-1"),
       timezone: String = "America/Los_Angeles"): ExternalUDFInitContext = {
     ExternalUDFInitContext(
@@ -267,10 +267,10 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
     val udf = pythonUDF(Seq(
       Literal(1),
       NamedArgumentExpression("named", Literal("value"))))
-    val workerSpec = PythonUDFWorkerSpecification.fromPythonFunction(
+    val workerSpec = PythonUDFWorkerSpecBuilder.build(
       udf.func,
       spark.sparkContext.getConf)
-    val external = PythonExternalUserDefinedFunction.fromPythonUDF(
+    val external = PythonExternalUDFAdapter.toExternalUDF(
       udf,
       workerSpec)
     val pythonTaskContext = Map(
@@ -286,7 +286,7 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
 
     assert(external.getClass === classOf[ExternalUserDefinedFunction])
     assert(init.getUdf.getName === "plus_one")
-    assert(init.getUdf.getFormat === "pyspark-udf-v2")
+    assert(init.getUdf.getFormat === "pyspark-udf-experimental")
     assert(init.getUdf.getEvalType === PythonEvalType.SQL_ARROW_BATCHED_UDF.toString)
     assert(init.getUdf.getInvocationId === udf.resultId.id)
     assert(init.getEnvironmentVariablesMap.isEmpty)
@@ -308,7 +308,7 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
       taskContext = nextTaskContext,
       dynamicConfig = pythonDynamicConfig,
       resourceDirectories = Map(
-        PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY ->
+        PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY ->
           "/var/resources/artifact-2"),
       timezone = "UTC"))
     assert(nextInit.getUdf === init.getUdf)
@@ -318,7 +318,7 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
     assert(nextInit.getTaskContextMap.get(
       ExternalUDFInitContext.IS_DRIVER_CONTEXT_KEY) === "true")
     assert(nextInit.getResourceDirectoriesMap.get(
-      PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY) ===
+      PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY) ===
       "/var/resources/artifact-2")
     assert(nextInit.getTimezone === "UTC")
 
@@ -359,7 +359,7 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
   }
 
   test("Python worker specification declares its generic session requirements") {
-    val spec = PythonUDFWorkerSpecification.fromPythonFunction(
+    val spec = PythonUDFWorkerSpecBuilder.build(
       pythonFunction(),
       spark.sparkContext.getConf)
     val session = spec.getSession
@@ -378,21 +378,21 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
       requirement.getIsRequired === !optionalKeys.contains(name)
     })
     assert(session.getRequiredResourceDirectoriesList.asScala.toSeq ===
-      Seq(PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY))
+      Seq(PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY))
     assert(session.getStaticConfigMap.isEmpty)
-    val defaultFunction = PythonExternalUserDefinedFunction.fromPythonUDF(
+    val defaultFunction = PythonExternalUDFAdapter.toExternalUDF(
       pythonUDF(Seq(Literal(1))),
       spec)
     val defaultInit = defaultFunction.buildInit(initContext(
       dynamicConfig = pythonDynamicConfig,
       resourceDirectories = Map(
-        PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY ->
+        PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY ->
           "/var/resources/artifact-default")))
     assert(defaultInit.getSessionConfMap.get(SQLConf.PYSPARK_BINARY_AS_BYTES.key) === "true")
     assert(!defaultInit.getSessionConfMap.containsKey(
       SQLConf.PYTHON_UDF_ARROW_CONCURRENCY_LEVEL.key))
     assert(defaultInit.getResourceDirectoriesMap.get(
-      PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY) ===
+      PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY) ===
       "/var/resources/artifact-default")
 
     val missingConfig = intercept[IllegalArgumentException] {
@@ -402,7 +402,7 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
 
     Seq(
       Map.empty[String, String],
-      Map(PythonUDFWorkerSpecification.ARTIFACTS_RESOURCE_DIRECTORY -> "")
+      Map(PythonUDFWorkerSpecBuilder.ARTIFACTS_RESOURCE_DIRECTORY -> "")
     ).foreach { resourceDirectories =>
       val error = intercept[IllegalArgumentException] {
         defaultFunction.buildInit(initContext(
@@ -417,23 +417,23 @@ class PythonExternalUserDefinedFunctionSuite extends QueryTest with SharedSparkS
   test("PySpark conversion rejects unsupported evaluation types") {
     val udf = pythonUDF(Seq(Literal(1)), PythonEvalType.SQL_BATCHED_UDF)
     val error = intercept[IllegalArgumentException] {
-      PythonExternalUserDefinedFunction.fromPythonUDF(
+      PythonExternalUDFAdapter.toExternalUDF(
         udf,
         UDFWorkerSpecification.getDefaultInstance)
     }
     assert(error.getMessage.contains("Unsupported Python external UDF eval type"))
   }
 
-  test("Python payload encoding has a stable versioned wire format") {
+  test("experimental Python payload encoding has an explicit wire version") {
     val payload = PythonUDFPayload.encode(pythonFunction(
       command = Array[Byte](1),
       pythonIncludes = Seq("x"),
       pythonVersion = "v"))
     val expectedPayload = hexBytes(
-      "505955440000000200000001010000000100000001780000000176")
+      "505955440000000100000001010000000100000001780000000176")
     assert(payload.sameElements(expectedPayload))
 
-    Seq(1, 3).foreach { version =>
+    Seq(0, 2).foreach { version =>
       val unsupportedVersion = payload.clone()
       ByteBuffer.wrap(unsupportedVersion).putInt(Integer.BYTES, version)
       assert(intercept[IllegalArgumentException] {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecificationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecBuilderSuite.scala
similarity index 96%
rename from sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecificationSuite.scala
rename to sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecBuilderSuite.scala
index bbd47741aba1c..af7044ee45ba7 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecificationSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/PythonUDFWorkerSpecBuilderSuite.scala
@@ -63,7 +63,7 @@ private class ConnectingTestDispatcher(spec: UDFWorkerSpecification)
 }
 
 /**
- * Tests that [[PythonUDFWorkerSpecification#fromPythonFunction]]
+ * Tests that [[PythonUDFWorkerSpecBuilder#build]]
  * produces a valid [[org.apache.spark.udf.worker.UDFWorkerSpecification]]
  * that can be used by a
  * [[org.apache.spark.udf.worker.core.WorkerDispatcher]]
@@ -80,7 +80,7 @@ private class ConnectingTestDispatcher(spec: UDFWorkerSpecification)
  * socket, it does not host a gRPC server, so a full gRPC session is not
  * exercised here.
  */
-class PythonUDFWorkerSpecificationSuite
+class PythonUDFWorkerSpecBuilderSuite
     extends SharedSparkSession {
 
   import IntegratedUDFTestUtils.{
@@ -142,7 +142,7 @@ class PythonUDFWorkerSpecificationSuite
     (moduleDir, moduleName)
   }
 
-  test("PythonUDFWorkerSpecification.fromPythonFunction" +
+  test("PythonUDFWorkerSpecBuilder.build" +
       " produces a spec that spawns a Python worker") {
     assume(isPySparkAvailable,
       "Python and PySpark must be available")
@@ -170,7 +170,7 @@ class PythonUDFWorkerSpecificationSuite
 
     // Build the spec via the function under test
     val workerSpec =
-      PythonUDFWorkerSpecification.fromPythonFunction(func, conf)
+      PythonUDFWorkerSpecBuilder.build(func, conf)
 
     // Verify the spec works end-to-end: the dispatcher spawns the Python
     // worker, waits for the socket, and opens a real UDS connection to it