peter-toth opened a new pull request, #58002: URL: https://github.com/apache/spark/pull/58002
### What changes were proposed in this pull request? `TransformExpression` is a case class holding a `BoundFunction`, so its compiler-generated equality compares that function by **instance**. `V2ExpressionUtils` binds the function afresh every time it converts a partitioning or ordering reported by a source, so two independently derived reports of the same `bucket(4, id)` are never equal. - `TransformExpression` gets `equals`/`hashCode`. Equality compares the function as `function.eq(other) || (canonicalName == canonicalName && resultType == resultType)`, then `numBucketsOpt` and `children` as before. `hashCode` is `Objects.hash(dataType, children, numBucketsOpt)`, cached. - `BoundFunction.canonicalName`'s javadoc now states that the name must be stable across `bind` calls, names both things Spark decides with it, and says that keeping the default is safe while two functions sharing a name is the case that can mislead Spark. - Binding warns when `canonicalName` disagrees with itself across two calls, so a connector on the default is diagnosable instead of silently slower. - `isSameFunction` gets the same instance short-circuit, making it reflexive. - `PlanMerger`'s `sameReportedExpressions` / `reportedOrderingSatisfies` -- added by SPARK-58549 as a local workaround for exactly this -- are deleted, reverting to `canonicalized ==` and `SortOrder.orderingSatisfies`. **Two design points worth stating up front.** *`hashCode` deliberately excludes the function.* Equality can now match two different instances, so hashing anything instance-specific would break the equals/hashCode contract, and hashing `canonicalName` would be unstable under the random-UUID default. `dataType` is safe because equality already implies both sides agree on it. *Why `equals` rather than `canonicalized`.* Overriding `canonicalized` is the narrower change and arguably the more idiomatic hook, but it needs a synthetic `BoundFunction` stand-in to hold the erased identity, and it does not reach `BatchScanExec.equals`, which compares the raw `keyGroupedPartitioning` field rather than the canonical form: | Consumer | Compares | Fixed by `equals` | by `canonicalized` | |---|---|---|---| | `ExpressionSet`, `semanticEquals`, `semanticHash` | canonical form | yes | yes | | `QueryPlan.canonicalized`, exchange/subquery reuse | canonical form | yes | yes | | `BatchScanExec.equals` | **raw field** | yes | no | Conceptually, two independently bound reports differ in no field anyone can observe -- they differ only because Spark called `bind()` twice -- so instance-distinctness is noise rather than a literal difference, much as `Literal(1) == Literal(1)` holds because the value identifies the expression. A viable alternative if reviewers prefer the convention: override `canonicalized` **and** change `BatchScanExec.equals` to compare `.map(_.canonicalized)`. **Relationship to `isSameFunction` / `isCompatible`.** Both remain, and not as workarounds -- they answer a different question: | | Question | Arguments | Used by | |---|---|---|---| | `equals` / `canonicalized` | same transform **and** same arguments? | compared | plan identity: dedup, reuse, canonicalization | | `isSameFunction` | same transform, arguments aside? | **ignored** | SPJ co-partitioning | | `isCompatible` / `reducers` | different transforms, reconcilable? | ignored | SPJ under `allowCompatibleTransforms` | A storage-partitioned join *must* ignore the arguments: it compares `bucket(4, left.id)` against `bucket(4, right.id)` and recovers the positions separately via `KeyedShuffleSpec.keyPositions`. No amount of correct `equals` provides that. Conversely plan identity has no such freedom, so the new equality is strictly stricter than `isSameFunction` -- a deliberate invariant, with a test for it, since relaxing plan identity to SPJ's notion could make Spark reuse a scan for a different computation. ### Why are the changes needed? Three places treat a reported key-grouped partitioning expression as an ordinary expression and so get the wrong answer today: - `PartitioningPreservingUnaryExecNode.projectKeyedPartitionings` puts them in an `ExpressionSet`, so deduplication silently fails, the per-position cross-product multiplies, and `.take(aliasCandidateLimit)` can truncate genuine alternatives. - `BatchScanExec.equals` compares `keyGroupedPartitioning`, so two identical bucketed scans miss plan reuse. - `DataSourceV2ScanRelation`'s canonical form does too, so `PlanMerger` cannot spot two identical subqueries over a bucketed table. The asymmetry behind all three: `canonicalName` is a documented obligation -- its default returns a random UUID *precisely* to force an override -- while `equals` on `BoundFunction` is required nowhere. So a connector meeting only the documented contract loses these optimizations: | Connector overrides | SPJ co-partitioning | Single-side kGP | Plan identity | |---|---|---|---| | `canonicalName` **and** `equals` | works | works | works | | `canonicalName` only -- the documented minimum | works | works | **broken** | | neither | broken | works | broken | Iceberg is in the first row today because `BaseScalarFunction` overrides `equals`/`hashCode` in terms of `canonicalName` -- but it sat in the middle row until apache/iceberg#9873 (2024), which is evidence the trap is real rather than theoretical. Spark should rest on the obligation it documents, not on a courtesy. Every symptom is an under-match -- a shuffle not skipped, a scan not reused -- so it is invisible and costs performance, never correctness. That also makes the fix safe: it adds matches that the connector has declared justified, keyed on a property SPJ already trusts for correctness. One claim I could not fully verify, and would ask reviewers to test: that nothing legitimately depends on distinguishing two bound instances of the same function via `==`. I searched Spark and found nothing; a third-party rule could differ. ### Does this PR introduce _any_ user-facing change? Yes, for DataSource V2 connectors that report partitioning or ordering via transforms. - A connector whose bound functions already compare equal semantically (Iceberg) sees **no change** -- for it this is a no-op. - A connector meeting only the documented contract now gets the deduplication, reuse and scan merging it was missing, so plans can change for such sources. - One case becomes stricter: a connector overriding `equals` by name alone, with a name coarse enough to be shared by functions of different result types, will stop seeing those treated as the same expression. That direction is correct -- they do not produce the same type -- and under-matching costs performance, not correctness. - A new warning is logged when a function's `canonicalName` is unstable. No new configuration, no API change; the `canonicalName` javadoc documents what Spark already relied on. ### How was this patch tested? New `TransformExpressionSuite` (catalyst, 11 tests) with fixtures that bind a fresh instance per call -- a fixture built on a singleton, as Spark's own `UnboundBucketFunction` is, would pass whatever the comparison did. It covers: two independent binds equal; reflexive under the UUID default; two UUID-default instances not equal; shared instance equal; same name with different `resultType` not equal; arguments and bucket count still discriminating; nested transforms; `ExpressionSet` deduplication; `equals` implies `isSameFunction`; and `isSameFunction` still ignoring arguments. Measured against a revert of the comparison to instance identity: 3 of the 11 fail (the equality ones -- the other 8 assert inequality or `isSameFunction` behaviour, pinning that this does not over-match), **and** SPARK-58549's `merge DSv2 scans reporting the same bucket transform partitioning` fails, which is what shows the deleted `PlanMerger` workaround is subsumed rather than merely unused. `build/sbt 'catalyst/testOnly *TransformExpressionSuite *DistributionSuite *V2ExpressionUtilsSuite'` -- 23 pass. `build/sbt 'sql/testOnly *KeyGroupedPartitioningSuite *ProjectedOrderingAndPartitioningSuite *MergeSubplansSuite *DSv2PlanMergingSuite *PlanMergingSuite'` -- 229 pass. `dev/lint-scala` clean. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) -- 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]
