Josh Rosen created SPARK-58154:
----------------------------------
Summary: Several expressions hold shared mutable state that is
unprotected by the stateful-expression mechanism, corrupting results under
concurrent evaluation
Key: SPARK-58154
URL: https://issues.apache.org/jira/browse/SPARK-58154
Project: Spark
Issue Type: Improvement
Components: SQL
Affects Versions: 4.2.0, 4.0.0, 4.1.0, 3.5.0, 4.3.0
Reporter: Josh Rosen
Several Catalyst expressions keep mutable evaluation state in instance fields
but are not marked {{{}stateful{}}}.
When the same expression _instance_ is evaluated from two threads at once, that
state is corrupted: queries silently return wrong results or, for some of the
expressions, fail with confusing errors.
I'm filing this single ticket to report an entire family of
conceptually-related expression statefulness bugs. Some of these may require
separate classes of fixes, in which case we can split sub-tasks or tickets from
this parent issue.
h3. Affected expressions found so far (this list may be incomplete)
# *{{{}NamedLambdaVariable{}}}, i.e. every higher-order function
({{{}transform{}}}, {{{}filter{}}}, {{{}aggregate{}}}, {{{}zip_with{}}}, ...) –
the most affected.* The lambda variable's current value lives in an
{{AtomicReference}} on the expression instance and is rewritten for every array
element in every row, so concurrent evaluation corrupts a large fraction of all
rows: lambda bodies compute over elements installed by the other thread, and
results contain values leaked from other rows (or possibly {_}other queries{_},
if subplans are shared).
Both interpreted evaluation and codegen go through the same
{{{}AtomicReference{}}}. Note that marking the _enclosing_ function stateful
does not help: {{TransformKeys}} is already {{stateful = true}} (for its map
builder), but fresh copies preserve non-stateful children, so its lambda
variables are still shared. Silent wrong results; reproduced on master (repro
below).
# *{{to_json}} ({{{}StructsToJson{}}}) – also reproduced on master.* Each
instance shares one {{{}CharArrayWriter{}}}/{{{}JacksonGenerator{}}}.
Concurrent evaluation of a shared instance usually fails with
{{{}com.fasterxml.jackson.core.JsonGenerationException: Can not start an
object, expecting field name{}}}, and occasionally returns silently corrupted
JSON instead (e.g. {{"a":0} for input \{"a":0}). }}Notably, this shared state
lives in an evaluator object that the expression embeds as a literal constant
in its RuntimeReplaceable replacement tree, so copying the expression tree
still reuses the one evaluator – this state is out of reach of the
stateful-copy mechanism even though the wrapping {{Invoke}} is marked stateful.
# *The other JSON/CSV/XML conversion functions* – {{{}to_csv{}}}, {{to_xml}}
(shared {{CharArrayWriter}} + generator, like {{{}to_json{}}}) and
{{{}from_json{}}}, {{{}from_csv{}}}, {{from_xml}} (shared parser objects, e.g.
{{{}FailureSafeParser{}}}) – keep similar shared evaluator state, though not
all via the literal-embedding pattern above; they are in scope for this audit.
# *{{format_number}} ({{{}FormatNumber{}}})* shares a
{{java.text.DecimalFormat}} (documented as not thread-safe) plus last-argument
caches in interpreted evaluation, so concurrent evaluation can produce
malformed output. Its codegen path keeps this state per generated-class
instance and appears unaffected.
# *{{{}regexp_replace{}}}, {{{}regexp_extract{}}}/{{{}regexp_extract_all{}}},
{{translate}}* cache the last pattern/replacement/dictionary in instance
fields. With literal pattern arguments, concurrent recomputation is idempotent
and harmless; with non-foldable (per-row) pattern arguments, the caches can
tear across threads (e.g. one thread's {{lastRegex}} paired with the other
thread's compiled {{{}Pattern{}}}), yielding wrong results.
h3. Reproduction (PySpark, current master, default configs)
Silent wrong results from {{transform}} when two threads collect sibling
DataFrames:
{code:python}
import threading
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
N, M, THREADS = 100, 1000, 2
vals = ",".join(f"({i})" for i in range(N))
# Inline VALUES becomes a LocalRelation, so ConvertToLocalRelation evaluates the
# projection on the driver during each derived query's optimization.
t = spark.sql(f"SELECT id FROM VALUES {vals} AS t(id)").select(
"id",
F.transform(
F.sequence(F.col("id") * M, F.col("id") * M + M - 1), lambda x: x + 1
).alias("t"),
)
# Sibling DataFrames: distinct Datasets/QueryExecutions whose logical plans
# share t's expression objects, including ONE NamedLambdaVariable instance.
branches = [t.select("id", "t") for _ in range(THREADS)]
results = [None] * THREADS
def work(k):
results[k] = branches[k].collect()
threads = [threading.Thread(target=work, args=(k,)) for k in range(THREADS)]
for th in threads: th.start()
for th in threads: th.join()
bad = sum(1 for res in results for r in res
if r["t"] != list(range(r["id"] * M + 1, r["id"] * M + M + 1)))
print("corrupted rows:", bad, "of", THREADS * N)
{code}
Observed on master (5.0.0-SNAPSHOT), five consecutive runs:
{noformat}
corrupted rows: 107 of 200
corrupted rows: 71 of 200
corrupted rows: 70 of 200
corrupted rows: 106 of 200
corrupted rows: 98 of 200
{noformat}
With {{THREADS = 1}} the same harness reports 0 corrupted rows on every run.
Scaling up (4 threads, 300 rows x 2000 elements) corrupted ~99% of rows in
every trial. Replacing the {{transform}} column with {{to_json(struct(id))}}
(2000 rows) in the same harness threw the {{JsonGenerationException}} above in
every run, and in some runs additionally returned silently corrupted JSON for
rows in the thread that did not fail.
h3. Paths that can lead to concurrent evaluation
There are multiple paths that can result in concurrent expression instance
evaluation:
* *Driver side (all currently-maintained versions):* DataFrames derived from a
common parent share the parent plan's expression instances, and
{{ConvertToLocalRelation}} evaluates projections over LocalRelations eagerly on
the driver, inside each derived query's optimization. As a result, the pattern
of building several DataFrames from one base DataFrame and collecting them from
a Python/Scala thread pool evaluates the shared instances concurrently. The
reproduction below uses this path.
* *Executor side (mainly 3.5.x and earlier, but still technically possible in
later versions):* before SPARK-44705 (4.0.0) made {{PythonRunner}}
single-threaded, operators upstream of a Python UDF ran in a separate "writer"
thread within each task. Optimizer rules that copy expression instances into
multiple plan positions – e.g. filter pushdown through a {{Project}} via
{{{}replaceAlias{}}}, or {{InferFiltersFromGenerate}} before SPARK-37392 (fixed
in 3.3.0) – could place the same instance on both sides of that thread
boundary. Spark 4.x closes that boundary under default configurations, though
SPARK-56642 (4.3.0/master) adds an opt-in pipelined mode
({{{}spark.python.udf.pipelined.enabled{}}}) that reintroduces a writer thread.
h3. Related tickets
Two earlier fixes hardened the fresh-copy mechanism for expressions that _are_
marked stateful – SPARK-49628 ({{{}ConstantFolding{}}} should copy stateful
expressions before evaluating) and SPARK-53275 (incorrect ordering in
interpreted mode when the sort order includes stateful expressions) – but the
expressions above carry no such marking, so the copy mechanism those fixes
invoke skips them. Also related: SPARK-25223 (closed, Won't Do) described the
{{NamedLambdaVariable}} value-passing design as fragile but did not report a
correctness bug; SPARK-44705 made {{PythonRunner}} single-threaded, which
incidentally narrowed (but did not close) the executor-side exposure;
SPARK-56642 adds an opt-in mode that reintroduces the writer thread.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
