spark-test commented on code in PR #60197:
URL: https://github.com/apache/spark/pull/60197#discussion_r4500852569
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala:
##########
@@ -117,7 +116,9 @@ case class LogicalRDD(
session :: originStats :: originConstraints :: Boolean.box(fromCheckpoint)
:: Nil
override def newInstance(): LogicalRDD.this.type = {
- val rewrite = Utils.toMap(output, output.map(_.newInstance()))
+ // Keyed by exprId, because the attributes in the partitioning, the
ordering, the stats and
+ // the constraints can differ from `output` in qualifier or nullability.
+ val rewrite = AttributeMap(output.zip(output.map(_.newInstance())))
Review Comment:
**Non-blocking (P2):** The same lookup miss survives one step earlier, where
a checkpoint's constraints and column stats are first built.
`LogicalRDD.fromDataset`, used by `checkpoint`, `localCheckpoint` and the
`foreachBatch` sink, re-keys them from the optimized plan onto the analyzed
output through `buildOutputAssocForRewrite`, which still returns a
full-equality `Map` (`.toMap`). `UnionLoopExec` (recursive CTEs) does the same
through `buildOutputAssocForRewriteWithNewOutput`. When the optimizer changes
the root expression IDs, a constraint attribute that differs from the optimized
output only in qualifier or nullability misses that map and keeps the source
plan's expression ID. No self-join or renewal is needed:
```scala
val df0 = spark.range(16).as("t").where($"t.id" > 5)
val cp = df0.select($"id".as("id")).localCheckpoint()
df0.join(cp, df0("id") === cp("id") + 1, "right_outer").where(cp("id") > 0)
```
`RemoveRedundantAliases` strips the root alias `t.id#0 AS id#1`, and
`RemoveNoopOperators` then drops the Project, so the optimized output is `id#0`
while the checkpoint's output is `id#1`. The constraint `t.id#0 > 5` misses
`Map(id#0 -> id#1)` and stays on `df0`'s column. With default configurations,
`EliminateOuterJoin` reads it as a null filter on the left side and turns the
join into an INNER join, so `Row(null, 15)` is dropped and only 9 of the
expected 10 rows come back. That is the PR's issue 2, reached through a join
with the source instead of a self-join. Column stats go through the same map,
so they also land on stale expression IDs and CBO cannot use them. (Traced
through the source at this head; not executed.)
**Recommended change:** Key the association that LogicalRDD's companion
helpers return by expression ID. After the existing positional name and type
checks pass, the helpers return an AttributeMap keyed by the optimized plan's
output attributes, so the stats and constraints rewrite moves every attribute
that shares an expression ID with an optimized output column onto the
LogicalRDD's own output, whatever its qualifier or nullability. Add a
checkpoint regression test, alongside the existing Dataset checkpoint tests,
for a checkpoint whose same-name root alias the optimizer removes, joined with
its source Dataset.
**Why this works:** The helpers already pair the optimized output with the
LogicalRDD output by position; only the lookup that follows compares full
attribute equality. Keying that lookup by expression ID gives checkpoint
creation and recursive-CTE iterations the same keying that renewal now uses.
Constraints and column stats can then no longer keep the source plan's
expression IDs, so a later join with the source cannot read them as predicates
over its own columns.
**Scope:** Expression-ID-keyed association in LogicalRDD's stats and
constraints rewrite helpers, plus a checkpoint regression test that joins a
checkpoint with its source.
**Compatibility:** When the optimized plan keeps the analyzed output's
expression IDs, the rewritten stats and constraints reference the same
expression IDs as today's; attributes that differ from the output only in
qualifier or nullability now take the output's. Outputs that fail the name and
type checks still drop stats and constraints with the existing warning.
**Risks:** Recursive CTE iterations can now keep constraints that previously
landed on stale or dangling expression IDs, so the optimizer may infer or prune
filters from them in later iterations. These constraints describe the previous
iteration's rows, which are exactly the new relation's rows, so results should
not change. Recursive CTE coverage should confirm this, and reverting the
helper keying rolls it back. Checkpoints and foreachBatch micro-batches whose
optimized root expression IDs differ from the analyzed output now carry their
constraints and column stats on their own columns, also in queries whose
results were already correct. The optimizer can then prune filters those
constraints imply, infer filters onto joined relations, and use the column
stats in cost-based estimates. These plans keep their results, and reverting
the helper keying rolls them back. Attributes that already had the right
expression ID only take the analyzed output's qualifier and nullability, which
canonical comparisons ignore.
**Constraints:** Keep the existing positional name and type checks and the
give-up path that drops stats and constraints with a warning when the outputs
do not line up. Change only how the validated association is keyed, so that an
output with repeated expression IDs is not newly rejected by the size check.
Keep the helpers' Map[Attribute, Attribute] result type, which
rewriteStatsAndConstraints and the existing constraint and stats tests in
DataFrameSuite and StatisticsCollectionSuite consume.
**Exclusions:**
- This repair leaves out remapping the executed plan's outputPartitioning
and outputOrdering in fromDataset onto the analyzed output when the optimizer
changes the root expression IDs. That is a separate pre-existing gap: those
fields are not re-keyed at all.
**Success:** When a checkpoint's stats and constraints come from an
optimized plan whose root expression IDs differ from the analyzed output, they
reference only the checkpoint's own output expression IDs, including attributes
that differ in qualifier or nullability. A right-outer join between such a
checkpoint and its source Dataset, filtered on the checkpoint's column, keeps
its null-extended rows. Recursive-CTE iterations re-key stats and constraints
onto the recursive reference's output the same way.
**Verification:**
- **Regression:** Checkpoint a Dataset whose optimized root output has
different expression IDs from its analyzed output and whose constraints
reference an output column through a differently qualified attribute, such as a
range aliased as t, filtered on t.id, then selecting id as id. Right-outer join
it with the source Dataset on an offset key, filter on the checkpoint's column,
and check that the unmatched checkpoint row is returned with a null left side.
- **Behavior:** When the optimized plan's root expression IDs differ from
the analyzed output, the checkpoint's constraints and column stats reference
only expression IDs in its own output, including attributes that differ from
the optimized output in qualifier or nullability.
- **Compatibility:** Existing coverage of LogicalRDD constraint and stats
re-keying, and of recursive CTE results, keeps its results.
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -1956,6 +1956,20 @@ class DatasetSuite extends SharedSparkSession
}
}
+ test("SPARK-61088: self-join of a checkpointed Dataset should not use the
partitioning " +
+ "and constraints of the other copy") {
+ withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false") {
+ // The partitioning and the constraints refer to `t.id`, while the
output is `s.id`.
+ val cp = spark.range(16).as("t").where($"t.id" >
5).repartition($"t.id").as("s")
Review Comment:
**Nit (P3):** This checkpoint ends in a `ShuffleExchangeExec`, which does
not override `outputOrdering`, so the checkpoint captures no ordering and the
`outputOrdering` rewrite in `newInstance` never sees a mismatched attribute.
The new code comment and the PR description both list the ordering among the
re-keyed fields, but a change that broke only that transform would still pass
both assertions here.
A stale ordering is its own wrong-result path. `spark.range(0, 16, 1,
1).as("s").localCheckpoint()` has no exchange, so AQE does not wrap it, and it
keeps `RangeExec`'s `SinglePartition` and its ordering on the unqualified `id`.
In `broadcast(cp.as("x")).join(cp.as("y"), $"x.id" % 2 === $"y.id" % 2)`
followed by `count(lit(1)).over(Window.partitionBy($"x.id"))`,
`SinglePartition` already satisfies the window's distribution, so only the
renewed copy's ordering decides whether a Sort is added. With a stale ordering,
the window counts each run of equal `x.id` separately instead of all 8 matching
rows.
Asserting that the renewed copy's `outputOrdering` refers only to its own
output, or checking a result over a single-partition checkpoint like this one,
would close the gap. A result check over a multi-partition range would not,
because the re-keyed partitioning already forces an Exchange and a Sort there.
**Verification:**
- **Regression:** With only the ordering re-keying reverted, the new case
fails, while the existing partitioning and constraints assertions still pass.
- **Behavior:** The renewed copy of a checkpoint that captured an ordering
on a qualifier-mismatched attribute reports that ordering on its own renewed
output column.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
