James Willis created SPARK-59009:
------------------------------------
Summary: InMemoryRelation.newInstance() does not remap
outputOrdering, causing NoSuchElementException during canonicalization
Key: SPARK-59009
URL: https://issues.apache.org/jira/browse/SPARK-59009
Project: Spark
Issue Type: Bug
Components: SQL
Affects Versions: 4.0.3, 4.1.2, 4.1.1, 4.2.0, 3.5.8, 4.0.2, 4.1.0, 3.5.9,
4.3.0, 4.1.3, 4.2.1, 4.1.4
Reporter: James Willis
h2. Symptom
A query over a cached ({{{}persist(){}}}) DataFrame that has a non-empty
{{outputOrdering}} fails at the first action with an internal error, if the
cached relation is referenced more than once (for example a CTE that is
referenced twice and then self-joined):
{noformat}
java.util.NoSuchElementException: key not found: tile#1L
at scala.collection.MapOps.default(Map.scala:289)
at
org.apache.spark.sql.catalyst.expressions.AttributeMap.apply(AttributeMap.scala:41)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation.$anonfun$withOutput$1(InMemoryRelation.scala:438)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation.withOutput(InMemoryRelation.scala:438)
at
org.apache.spark.sql.execution.columnar.InMemoryRelation.doCanonicalize(InMemoryRelation.scala:406)
...
at
org.apache.spark.sql.catalyst.plans.QueryPlan.canonicalized(QueryPlan.scala:633)
at
org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanExec.createNonResultQueryStages(AdaptiveSparkPlanExec.scala:589)
{noformat}
The error surfaces at {{.show()}} / {{.collect()}} and gives the user nothing
actionable, since the query itself is well formed.
h2. Reproduction
Deterministic, no external data:
{code:python}
spark.range(0, 20).selectExpr("id", "id % 3 AS
tile").createOrReplaceTempView("a")
# a persisted DataFrame with a non-empty outputOrdering
b = spark.sql("SELECT id, tile FROM a ORDER BY tile, id")
b.persist()
b.count()
b.createOrReplaceTempView("b")
# window-rank it, then self-join the ranked result
spark.sql("""
WITH r AS (SELECT *, row_number() OVER (PARTITION BY tile ORDER BY id DESC)
AS rn FROM b)
SELECT x.id AS p, y.id AS q
FROM r x JOIN r y ON x.tile = y.tile AND x.rn = 1 AND y.rn = 2
""").show()
{code}
Verified failing on 4.0.4, 4.1.3 and 4.2.0. Passes on 4.0.1.
The invariant break can also be shown directly, without running a query:
{code:scala}
val df = spark.range(0, 20).selectExpr("id", "id % 3 AS tile").orderBy("tile",
"id")
df.persist(); df.count()
val ir =
spark.sharedState.cacheManager.lookupCachedData(df).get.cachedRepresentation
ir.output // id#0L, tile#1L
ir.outputOrdering // tile#1L ASC, id#0L ASC <- consistent
val ni = ir.newInstance()
ni.output // id#68L, tile#69L <- refreshed
ni.outputOrdering // tile#1L ASC, id#0L ASC <- NOT refreshed
ni.canonicalized // throws NoSuchElementException
{code}
h2. Root cause
{{InMemoryRelation}} carries an implicit invariant: {{outputOrdering}} may only
reference attributes present in {{{}output{}}}. {{newInstance()}} breaks it,
because it gives {{output}} fresh exprIds but passes {{outputOrdering}} through
unchanged:
{code:scala}
override def newInstance(): this.type = {
InMemoryRelation(
output.map(_.newInstance()), // fresh exprIds
cacheBuilder,
outputOrdering, // NOT remapped -- still the old exprIds
statsOfPlanToCache).asInstanceOf[this.type]
}
{code}
This was harmless until SPARK-53738, which rewired {{doCanonicalize}} through
{{withOutput}} and made {{withOutput}} remap the ordering with a strict
{{AttributeMap}} lookup:
{code:scala}
override def doCanonicalize(): logical.LogicalPlan =
withOutput(output.map(QueryPlan.normalizeExpressions(_, output)))
def withOutput(newOutput: Seq[Attribute]): InMemoryRelation = {
val map = AttributeMap(output.zip(newOutput))
val newOutputOrdering = outputOrdering
.map(_.transform { case a: Attribute => map(a) }) // throws on a missing
key
.asInstanceOf[Seq[SortOrder]]
InMemoryRelation(newOutput, cacheBuilder, newOutputOrdering,
statsOfPlanToCache)
}
{code}
So any {{InMemoryRelation}} that has been through {{newInstance()}} now fails
the moment anything canonicalizes it.
How {{newInstance()}} comes to be called on a cached relation: cache
substitution ({{{}CacheManager.useCachedData{}}}) runs on the analyzed plan,
before the optimizer. {{InlineCTE}} then inlines a CTE that is referenced more
than once, and to obtain a fresh-exprId copy it runs {{DeduplicateRelations}}
over a synthetic self-join (InlineCTE.scala, the {{case ref: CTERelationRef}}
branch). By that point the plan already contains {{{}InMemoryRelation{}}},
which is a {{{}MultiInstanceRelation{}}}, so {{DeduplicateRelations}} calls
{{newInstance()}} on it.
This is also why a plain self-join of a cached temp view does *not* reproduce:
that deduplication happens in the analyzer, before caching is substituted, so
the {{InMemoryRelation}} is never passed through {{{}newInstance(){}}}.
h2. Affected versions
The {{withOutput}} change is present in 3.5.8+, 4.0.2+, and every branch from
branch-4.0 through branch-4.3, branch-4.x and master (5.0.0-SNAPSHOT). 3.5.7,
4.0.0 and 4.0.1 are unaffected. {{newInstance()}} is still unpatched
everywhere, so no released version since 4.0.1 contains a fix.
h2. Suggested fix
Route {{newInstance()}} through {{withOutput}} so the ordering is remapped onto
the fresh exprIds. This preserves the ordering rather than dropping it:
{code:scala}
override def newInstance(): this.type =
withOutput(output.map(_.newInstance())).asInstanceOf[this.type]
{code}
And make {{withOutput}} defensive, since {{outputOrdering}} is only an
optimization hint and no cache hint should be able to hard-fail a query:
{code:scala}
val newOutputOrdering = if
(outputOrdering.forall(_.references.forall(map.contains))) {
outputOrdering.map(_.transform { case a: Attribute => map(a)
}).asInstanceOf[Seq[SortOrder]]
} else {
Nil
}
{code}
Note that {{withOutput}} also silently truncates when {{newOutput}} is shorter
than {{{}output{}}}, because of the {{{}output.zip(newOutput){}}}. The guard
above covers that case too.
h2. Workarounds
{{localCheckpoint()}} or {{checkpoint()}} instead of {{{}persist(){}}}, or
breaking the ordering before caching (for example {{repartition()}} after the
sort). There is no configuration-level workaround: disabling AQE
({{{}spark.sql.adaptive.enabled=false{}}}), disabling broadcast joins
({{{}spark.sql.autoBroadcastJoinThreshold=-1{}}}), and excluding {{InlineCTE}}
via {{spark.sql.optimizer.excludedRules}} all still fail.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
