What changes were proposed in this pull request?

Python UDFs on Spark Connect run in worker processes whose environment is always empty: the Connect
planner builds every Python function with no environment variables. This lets a session carry an
environment for its Python workers through session configurations under a reserved prefix, one
configuration per variable.

spark.conf.set("spark.pythonWorkerEnv.MY_SETTING", "abc")

@udf("string")
def f(_):
    import os
    return os.environ["MY_SETTING"]   # "abc"; previously a KeyError

Changes, all confined to sql/connect/server apart from four new error sub-conditions:

• PythonWorkerEnvironment (new): reads the environment from the session's configurations,
  validates it, and hands out a fresh mutable copy for a single Python function.
• SparkConnectPlanner.transformPythonFunction populates SimplePythonFunction.envVars from it.
  Every Python function family built at that site therefore receives the environment. Covered by
  tests: scalar Python UDFs, Arrow-batched UDFs, scalar pandas UDFs and their iterator variant,
  mapInPandas and mapInArrow. Reaching the same site by construction but not separately tested
  here: grouped-map, cogrouped-map, stateful pandas functions, streaming foreach /
  foreachBatch callbacks, and Python listeners.
• SparkConnectConfigHandler.handleSet validates a write under the prefix before storing it, so a
  malformed variable set through the config RPC is refused instead of being kept in the session.
  The check and the write it validated happen as one unit, under the monitor that guards the
  session configurations.
• The prefix is documented under a new Session Configuration heading in the Spark Connect section
  of docs/configuration.md.
• The session plan cache is keyed on the environment in addition to the relation.
• Three internal, cluster-level configurations bound the environment: at most 100 variables, names
  at most 512 characters, and 128 KiB total measured as the sum of the UTF-8 lengths of every name
  and value. Zero accepts no user-provided environment at all; a negative value is rejected.

Design notes a reviewer may want:

• The configurations are the authoritative session state. No second copy is maintained as
  session state, so the environment follows the session wherever ordinary session configurations
  follow it: it survives reattach and retry with no code, and SQLConf.clone() carries it into a
  session created by cloneSession, while newSession correctly starts without one. There is a
  test for each.
• One snapshot per request. A request reads the environment once and uses that snapshot for the
  cache lookup, for building its Python functions, and for the cache insertion. Re-reading would
  race with a concurrent configuration write and could store a plan built with one environment
  under the key of another.
• The cache key holds the request's snapshot. A cached plan bakes the environment into every
  Python function it contains, so an entry is only reusable by a request carrying the same
  environment. Note the key is therefore as large as the environment. A write through the config
  RPC is bounded by the limits before it is stored, but one through SQL SET is not, so the limits
  bound what reaches a worker rather than what the session and its cache can be made to hold.
• A fresh mutable copy per function is required. BasePythonRunner takes the map by reference
  and writes its own entries into it before launching a worker, so a shared map would leak entries
  between functions and an immutable one would fail the assignment.
• Validation happens in two places, and both are needed. A write through the Connect config
  RPC is validated before it is stored, against the environment the write would produce, so a
  malformed variable is refused at the call that set it and the session is never left carrying an
  environment its queries cannot use. That covers spark.conf.set and the options
  SparkSession.builder.config applies. It cannot be the only check: SQL SET reaches the session
  configurations through SetCommand, and application-level configurations are merged into a new
  session, neither of which passes through the RPC. So the environment is validated again when a
  Python function is built, which every write path does reach, and that is what guarantees an
  invalid environment cannot reach a worker. The remaining cost of the second layer is that an
  environment installed by one of those paths stays in the session until the user corrects it,
  failing only the queries that would install it rather than every query in the session.
  Covering those paths at write time instead would mean a prefix-validator hook in
  SQLConf.setConfString, which lives in sql/catalyst and cannot see the Connect server.
• The write-time check and its write are atomic. They are performed under
  SQLConf.settings, the monitor every writer of the session configurations takes:
  setConfString and getAllConfs both hold it, and SQLConf.setConf(props) holds it across a
  compound write for the same reason. Without it two concurrent writers could each validate against
  the environment as it was before the other's write and jointly exceed a limit that neither write
  appeared to break. This closes the window between the check and the write, including against a
  concurrent SQL SET; it does not make SQL SET itself validated.
• A rejection never carries the value, on any path. The silent path reports a refusal as a
  warning, and that warning outlives the response: the Scala client logs it and the Python client
  raises it. It therefore names only the configuration key. Reads through this handler are already
  redacted through spark.redaction.regex, so a write path that echoed values back would have
  contradicted the handler's own policy.
• A removal is deliberately not validated. It can only shrink the environment, and it is how a
  session recovers from an environment that one of the unchecked write paths left invalid.
• A value containing NUL is rejected. A process environment cannot carry it, and the JDK's own
  rejection embeds the offending value in its message, so this has to be caught before a worker
  launch is attempted.
• Names are preserved case-sensitively. On a case-sensitive operating system FOO and foo are
  therefore distinct; Windows process environments are case-insensitive, so what a worker observes
  there is the platform's business.
• The accepted name pattern is deliberately stricter than the OS requires. POSIX permits any byte
  except = and NUL, and container platforms accept their own broader sets, but a name outside
  [A-Za-z_][A-Za-z0-9_]* cannot be referenced portably from a shell. It is a portability policy,
  not a description of what a process environment can hold.
• The name pattern is checked with a whole-string match rather than a search. An anchored pattern
  that is searched for would accept a name with a trailing newline, since $ also matches before a
  terminating line break.
• Rejections reuse the existing INVALID_SPARK_CONFIG condition rather than adding a new top-level
  one. A message may name a variable but never carries its value, and a name is truncated and has
  its control characters escaped, so a rejection cannot forge log lines. Note the name itself is
  user-chosen, so it is only as safe as what the user put in it.
• An empty value is accepted (FOO= in a shell). A null value needs no handling: SQLConf
  rejects one on the way in, so a config request with an absent value fails rather than storing
  null.

Python UDTFs (transformPythonTableFunction) and Python data sources (transformPythonDataSource)
have their own construction sites and keep receiving an empty environment; they are follow-ups.

Not addressed here, and worth a reviewer's attention: a user can set a name that Spark's own worker
protocol uses. Spark wins every contested write — BasePythonRunner takes this map by reference and
overwrites its own keys in it, and PythonWorkerFactory applies the map before setting its own — so
a user value never displaces one that Spark writes, PYTHON_WORKER_FACTORY_SECRET and
PYTHONUNBUFFERED included.

The gap is the variables Spark writes only under a condition. SPARK_REUSE_WORKER,
SPARK_HIDE_TRACEBACK, SPARK_SIMPLIFIED_TRACEBACK, SPARK_TRACEBACK_WITH_LOCALS,
SPARK_PIPELINED_UDF, PYSPARK_EXECUTOR_MEMORY_MB, PYTHON_FAULTHANDLER_DIR,
PYSPARK_SPARK_SESSION_UUID, OMP_NUM_THREADS and the transport variables are not cleared when
their condition is false, so a user value survives and Spark's own code then reads it back as though
Spark had set it: PYTHON_FAULTHANDLER_DIR becomes a File, and PYSPARK_SPARK_SESSION_UUID
selects where worker logs are routed. Spark's intent in those branches is that the variable be
unset, which is currently expressed as not writing it rather than as removing it.

PYTHONPATH is a separate case: it is merged rather than replaced, after Spark's own entries, so a
user can add importable module paths to a worker, though not shadow pyspark's own modules. Clearing
the conditional names when their condition is false, and deciding whether merging PYTHONPATH is
wanted, are left to a follow-up.

Why are the changes needed?

Code that reads os.environ behaves differently inside a Python UDF than outside it, and a Spark
Connect client has no way to influence it. A user can set a value, read it successfully from driver
code, and get a KeyError for the same name inside a UDF.

This is also the gap that blocks moving existing workloads onto Spark Connect: on classic compute an
executor environment can be configured for the application through spark.executorEnv.*, but that
is application-scoped and set before the context starts, so it has no session-scoped equivalent a
Connect client can use.

Does this PR introduce any user-facing change?

Yes. Session configurations under spark.pythonWorkerEnv. are now read and installed in the
environment of the Python worker processes that run the session's Python functions, so os.environ
inside a Python UDF can see them. Previously these configurations had no effect, and the worker
environment was always empty.

Setting a malformed or oversized environment now fails, with
INVALID_SPARK_CONFIG.INVALID_PYTHON_WORKER_ENV_VAR_NAME,
INVALID_SPARK_CONFIG.INVALID_PYTHON_WORKER_ENV_VAR_VALUE,
INVALID_SPARK_CONFIG.PYTHON_WORKER_ENV_TOO_MANY_VARIABLES, or
INVALID_SPARK_CONFIG.PYTHON_WORKER_ENV_TOO_LARGE. Through spark.conf.set it fails at the call
that set it, and nothing is stored. Through SQL SET or an application-level configuration it is
stored and fails the queries that would install it in a worker.

SparkSession.builder.config applies its options silently, so a malformed variable set there is
reported as a warning and not stored rather than failing session creation, which is how the builder
already treats an option it cannot apply.

The warning that a silent configuration request returns for a write it could not apply no longer
repeats the rejected value, for any configuration key, only the key and the reason. The prefix is
also now documented in docs/configuration.md.

The three new bounding configurations are internal.

How was this patch tested?

PythonWorkerEnvironmentSuite (new, 36 tests):

• Reading: variables under the prefix, configurations outside the prefix ignored, an empty value,
  case sensitivity, no configurations at all, and that a null value cannot be installed.
• Validation: a malformed name (including a trailing newline and an empty name), a name over the
  length limit with the name bounded in the message, a value containing NUL, more variables than
  the limit, exactly the limit, a total size over the limit, and a total size that exceeds the
  limit only when counted in UTF-8 bytes rather than characters.
• Message safety: a name carrying newlines, tabs, DEL and an ANSI escape is escaped, and the
  message carries no control characters.
• Limits: each of the three is exercised at a non-default value, zero accepts nothing, and a
  negative value is rejected by the configuration itself.
• Plan cache: an entry is keyed on the snapshot the plan was built with; a change to the
  configurations midway through planning does not mis-key the entry, and the plan built under the
  old environment is not reused afterwards; five successive environments produce five entries; an
  oversized environment still does not stop an ordinary query being planned and cached.
• Session lifecycle also covers the Connect cloneSession path through the session manager, not
  only SparkSession.cloneSession.
• Delivery: every scalar family and mapInPandas / mapInArrow receive the environment; an empty
  one when nothing is set; each function gets an independent mutable copy; an invalid environment
  fails planning of a Python function but not of a plan without one.

SparkConnectConfigHandlerSuite (9 tests added): a valid variable is stored; an invalid name and a
value carrying NUL are refused and not stored; a limit on the collection is enforced against the
environment the write would produce, while an already accepted write stays; a silent request reports
a refusal as a warning instead of failing, and that warning carries the key but not the rejected
value; two writers racing at a barrier against a limit of one variable end with exactly one write
accepted and the limit intact; Unset is not validated, so a session can leave an environment
installed through a path the RPC does not see; and a key outside the reserved prefix is not
validated as a variable name.

SparkConnectPythonWorkerEnvTests (new, end-to-end through a real Connect client and a real Python
worker): a UDF reads the value from os.environ; an unset name is not visible; an update is picked
up; unset removes it; an empty value arrives as empty; a platform-owned variable
(PYTHONUNBUFFERED) still wins; an invalid name and a NUL value fail spark.conf.set itself
without printing the value and leave nothing stored; the same invalid name installed through SQL
SET instead is stored and fails the query, which is what exercises the second layer; and
mapInPandas sees the environment.

SparkConnectSessionHolderSuite was updated for the new plan cache key and its plan cache tests
still pass. SparkThrowableSuite passes with the new error sub-conditions. The end-to-end suite was
run against a real Connect server and real Python workers: 11 tests, all passing.

build/sbt "connect/testOnly *PythonWorkerEnvironmentSuite *SparkConnectSessionHolderSuite"
build/sbt "core/testOnly *SparkThrowableSuite"
python/run-tests --testnames pyspark.sql.tests.connect.test_connect_python_worker_env

connect/scalastyle, connect/Test/scalastyle, and scalafmt (with CI's changedOnly=false) are
clean.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5)