peter-toth commented on code in PR #55885:
URL: https://github.com/apache/spark/pull/55885#discussion_r4122106780
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -52,23 +87,46 @@ case class TransformFunctionId(canonicalName: String,
numBucketsOpt: Option[Int]
case class TransformExpression(
function: BoundFunction,
children: Seq[Expression],
- numBucketsOpt: Option[Int] = None,
- reducedWith: Option[TransformFunctionId] = None) extends Expression {
+ reducedWith: Option[TransformFunctionId] = None) extends Expression with
Logging {
override def nullable: Boolean = true
/** Drops the trailing `reducedWith` when it is unset, so the ordinary form
is unchanged. */
override protected def stringArgs: Iterator[Any] =
if (reducedWith.isEmpty) super.stringArgs.take(productArity - 1) else
super.stringArgs
- /** The identity of this expression's transform function. */
- lazy val functionId: TransformFunctionId =
- TransformFunctionId(function.canonicalName(), numBucketsOpt)
+ /**
+ * Extract literal children (constant parameters) from this transform. These
are constant
+ * arguments like the bucket count in `bucket(n, col)` or the width in
`truncate(col, width)`.
+ * Positions are dropped, so this is for handing parameters to a connector
reducer, NOT for
+ * identity -- see [[functionId]] and [[TransformFunctionId.argumentShapes]].
+ */
+ private lazy val literalChildren: Seq[Literal] =
+ children.collect { case l: Literal => l }
/**
- * Whether this [[TransformExpression]] has the same semantics as `other`.
- * For instance, `bucket(32, c)` is equal to `bucket(32, d)`, but not to
`bucket(16, d)` or
- * `year(c)`.
+ * This transform's identity, or None if some argument is not a literal, a
nested transform or a
+ * column reference (e.g. `c + 1`); such a transform is not the same as any
other.
+ */
+ lazy val functionId: Option[TransformFunctionId] = {
+ import TransformFunctionId.ArgumentShape
+ val shapes = children.map {
+ case l: Literal => Some(ArgumentShape.Param(l))
+ case t: TransformExpression => t.functionId.map(ArgumentShape.Nested(_))
+ case c if TransformExpression.isColumnRef(c) =>
Some(ArgumentShape.Column)
Review Comment:
**Finding 4.** This arm gives up the identity when an argument is not a
literal, a nested transform or a plain column ref, so `functionId` is `None`
and `isSameFunction` answers false for such a transform against itself.
`ShuffleSpec.isCompatibleWith` documents the opposite at
`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:1586`:
"Note that Spark assumes this to be reflexive, symmetric and transitive". The
new `TransformExpressionSuite` test pins the non-reflexivity ("never the same,
not even as itself").
It is reachable under `v2BucketingShuffleEnabled`.
`KeyedShuffleSpec.createPartitioning` builds the shuffled side's expression
from the *other* side's clustering key, so a join `ON t1.id = t2.b + 1` over a
`bucket(4, id)` table yields `bucket(4, b + 1)` on the shuffled side, paired
with `bucket(4, id)` on the keyed one. `keyPositions` reconciles them, one
reference each, and the two are co-located by construction, since
`createPartitioning` built the expression for exactly that. Master's
`isSameFunction` said true for that pair; this head says false.
The root cause is that identity is being asked two questions. "Which
transform is this" is identity's; "may SPJ reason about these argument shapes"
is `KeyedPartitioning.supportsExpressions`', which already asks it and is the
load-bearing gate on the scan. Once identity stops re-asking it, the whole
thing gets shorter:
```scala
lazy val functionId: TransformFunctionId = {
import TransformFunctionId.ArgumentShape
TransformFunctionId(function.canonicalName(), children.map {
case l: Literal => ArgumentShape.Param(l)
case t: TransformExpression => ArgumentShape.Nested(t.functionId)
case _ => ArgumentShape.Slot
})
}
```
with `Column` renamed to `Slot`, since it now covers any non-parameter
argument. What that discriminates is unchanged where it matters:
- `truncate(id, 2)` and `truncate(2, store_id)` stay apart, the `Param`
positions differ
- `bucket(4, years(c))` and `bucket(4, days(c))` stay apart, `Nested` is
still matched above
- `bucket(32, c)` and `bucket(16, c)` stay apart
and the knock-on simplifications fall out:
- `isSameFunction` becomes `functionId == other.functionId`
- `reducedKeySpace` goes back from the for-comprehension to
`reducedWith.map(partner => Set(functionId, partner))`
- `reducedTogetherWith` loses its `SparkException.internalError` arm, which
its own comment already called unreachable, so it is a one-liner again
- the `SparkException` import goes
Net `-17` lines in `TransformExpression.scala`.
On whether the wider `Slot` is safe. Identity is column-blind by
construction, and the safety comes from `keyPositions` plus
`supportsExpressions`, not from identity. For two non-column-slot transforms to
be compared at all, both have to be `createPartitioning` products, since the
scan gate refuses such a shape, and those are built from one spec onto one key
set, so "same function" is the right answer there. I could not construct a case
where the wider slot co-locates rows that do not belong together; if you see
one, that settles it the other way.
I ran this on this head. Green: `TransformExpressionSuite`,
`ShuffleSpecSuite` (41), then `KeyGroupedPartitioningSuite`,
`ProjectedOrderingAndPartitioningSuite`, `GroupPartitionsExecSuite`,
`EnsureRequirementsSuite`, `PlannerSuite` (438). Four assertions had to flip,
and three of them are the ones that enshrine the non-reflexivity:
- `TransformExpressionSuite`, "a transform with a non-reference argument has
no identity" -- becomes "a non-parameter argument is a slot, whatever its shape"
- the `plusOne` assertion in "transform identity is per-argument-position"
- the `add` assertion in `KeyGroupedPartitioningSuite`'s "isSameFunction is
a structural per-position walk"
- the `shapes.forall(_.functionId.isDefined)` fixture guard in
"isSameFunction and hasSameReducedKeys agree on every pair", which can instead
gain a non-reference shape to the fixture list, since the agreement now holds
over those too
Pushed both changes on top of `18f6f8f` in case cherry-picking is easier
than redoing them:
[`SPARK-50593-simplify-identity-and-alias-projection`](https://github.com/peter-toth/spark/compare/18f6f8fbe132eb2750e8a3ed1a11f2315ad2b083...peter-toth:spark:SPARK-50593-simplify-identity-and-alias-projection)
on my fork. Two independent commits, so you can take either on its own:
- [`4c51b96`](https://github.com/peter-toth/spark/commit/4c51b96c0b1) --
this finding, the total `functionId` plus the four test assertions that flip
- [`e93ae9b`](https://github.com/peter-toth/spark/commit/e93ae9bd32a) --
finding 1, the one-clause `projectExpression` rule,
`projectPartitionExpression` deleted, and the ordering assertion added to
`ProjectedOrderingAndPartitioningSuite`
75 insertions and 78 deletions over 6 files. Same content I ran the suites
against; take it, adapt it or ignore it.
--
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]
