[
https://issues.apache.org/jira/browse/SPARK-58154?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Josh Rosen updated SPARK-58154:
-------------------------------
Issue Type: Bug (was: Improvement)
> 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: Bug
> Components: SQL
> Affects Versions: 3.5.0, 4.1.0, 4.0.0, 4.2.0, 4.3.0
> Reporter: Josh Rosen
> Priority: Major
> Labels: correctness
>
> 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]
