This is an automated email from the ASF dual-hosted git repository.
peter-toth pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 3c1903e8ea6a [SPARK-58769][SQL] Document the BoundFunction equality
contract and drop the name-based transform comparison
3c1903e8ea6a is described below
commit 3c1903e8ea6af150891b0fe3b1604c4bcc2772cd
Author: Peter Toth <[email protected]>
AuthorDate: Fri Aug 21 15:23:11 2026 +0200
[SPARK-58769][SQL] Document the BoundFunction equality contract and drop
the name-based transform comparison
### What changes were proposed in this pull request?
Spark compares partition transform expressions with ordinary expression
equality, which reaches `BoundFunction.equals`. It binds a function afresh
every time it converts a partitioning or ordering reported by a source, so two
reports of one transform -- from two scans of the same table, or from the
partitioning and the ordering of a single scan -- hold two bound instances. A
function that does not implement `equals`/`hashCode` therefore yields
expressions that never compare equal acro [...]
- **`BoundFunction` gets an `equals`/`hashCode` contract** (both redeclared
on the interface): implementations SHOULD override them, comparing whatever
state affects behaviour; comparing `canonicalName` alone is not necessarily
enough; whatever is compared must be stable across `bind` calls; `hashCode`
must agree, since these expressions land in hash-based collections. Two
functions that can produce different values must **not** compare equal -- the
same comparison decides whether two [...]
- **The relationship between the two comparisons is stated**: `equals` is
the finer of the two, so two functions that compare equal must return the same
`canonicalName`, while the same name does not make them equal. An
implementation overriding `equals` should override `canonicalName` too, instead
of claiming two instances are the same expression but not the same partition
function -- which would cost it exactly the co-partitioning the name exists to
enable.
- **`canonicalName`'s javadoc is scoped** to the question it actually
answers -- whether two transforms are the same partition function -- says it is
deliberately coarser than `equals`, and leads with the one requirement here
that guards results rather than performance: two functions that partition data
differently must not return the same name.
- **`PlanMerger`'s `sameReportedExpressions` / `reportedOrderingSatisfies`
are removed**, so it compares reported transforms the way every other rule does.
**Why not derive the identity in Spark instead?** Because it cannot be
derived. `canonicalName` says nothing about `inputTypes`, `resultType`,
`isResultNullable`, `isDeterministic`, a `ReducibleFunction`'s reducers, or
whatever else an implementation holds. Adding the function's class separates
two different classes, but not two instances of **one** class that captured
different state when they were bound -- a width, a precision, a timezone, a
snapshot id -- and `BoundFunction` is an [...]
**Relationship to `isSameFunction` / `isCompatible`.** They are untouched,
and answer a different question:
| | Question | Arguments | Used by |
|---|---|---|---|
| `equals` (via the connector) | same transform **and** same arguments? |
compared by Spark | plan identity: dedup, reuse, canonicalization |
| `isSameFunction` | same transform, arguments aside? | children
**ignored**, bucket count compared | SPJ co-partitioning |
| `isCompatible` / `reducers` | different transforms, reconcilable? |
children 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` -- so no amount of correct
`equals` replaces it. Note the division of labour in the other direction too: a
coarse connector-side `equals` does not have to carry the whole identity,
because Spark compares the transform's arguments and bucket count itself.
### Why are the changes needed?
Every place that compares a reported key-grouped partitioning or ordering
treats it as an ordinary expression, so all of them miss matches for a
connector without `equals`:
- `UnionExec.comparePartitioning` compares the children's partition
expressions with `semanticEquals`, *after* `prepareOutputPartitioning` has
remapped them onto the first child's attributes. For a union of two scans of
one bucketed table the expressions are then identical except for the function
instance, so the partitioning merge is declined and the union loses its
`KeyedPartitioning`.
- `GroupPartitionsExec.outputOrdering` keeps the sort orders whose
expression is a partition key expression, via
`ExpressionSet(keyedPartitionings.flatMap(_.expressions)).contains(order.child)`.
The reported ordering and the reported partitioning come from two separate
binds, so the reported ordering is dropped.
- `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.
SPARK-58549 made `PlanMerger` the one exception, comparing reported
transforms by `isSameFunction` -- canonical name, arguments ignored -- instead
of by equality. The inconsistency is sharpest inside `PlanMerger` itself:
`checkIdenticalPlans` asks `newPlan.canonicalized == cachedPlan.canonicalized`,
which for a connector without `equals` misses the very pair the workaround then
merged the expensive way. So that connector was denied the cheap identical-plan
reuse and granted the costly [...]
Worth being explicit about what the removal is *not* justified by: at this
one site the deleted comparison could not actually conflate two different
functions. `tryMergeScanRelations` gates the merge on
`np.relation.canonicalized == cp.relation.canonicalized` -- the same table,
catalog, options and identifier -- and both reports come from binding that one
`UnboundFunction` with the same input types, so the two instances differ only
if `bind` is non-deterministic. The argument for remo [...]
Iceberg's `BaseScalarFunction` implements `equals`/`hashCode` in terms of
`canonicalName` and is therefore unaffected; it added them in
apache/iceberg#9873, which is evidence a connector can be caught by this in
practice. What was missing is that Spark never said it depended on them.
### Does this PR introduce _any_ user-facing change?
No.
- **No released behaviour changes.** The only behaviour this PR alters is
the scan merge SPARK-58549 added, and that is in no release: it is on `master`
and on `branch-4.x` at 4.4.0-SNAPSHOT, which are the two branches this PR
targets. Every release that will ever contain the workaround will also contain
this change, so no connector loses a merge it ever shipped with. A connector
without `equals`/`hashCode` keeps missing the four matches above exactly as it
does today.
- **No API change.** The interface now declares `equals` and `hashCode`,
but both are inherited from `Object`, so every existing implementation already
satisfies them, nothing needs recompiling, and MiMa reports nothing. No new
configuration.
- **Nothing else moves.** No plan changes for any other connector, no
change to `isSameFunction`, `isCompatible` or storage-partitioned join, and no
change for a connector that implements the contract (Iceberg).
- What does change is the published `BoundFunction` javadoc, which
documents an obligation Spark already relied on.
### How was this patch tested?
New `TransformExpressionSuite` (catalyst, 3 tests) for what the contract
means: expression equality follows the function's own `equals` -- a
non-comparable function's two binds do not compare equal, a shared instance
does, and a comparable one deduplicates in an `ExpressionSet`; the two
comparisons agree for a function that follows the contract, while still
differing in what they take into account (`isSameFunction` ignores the
arguments, equality does not); and a coarse `equals` does [...]
`MergeSubplansSuite` is where the behaviour change is covered, in both
directions. Its `TestBucketFunction` now implements the contract, which is what
keeps SPARK-58549's `merge DSv2 scans reporting the same bucket transform
partitioning` passing with the workaround gone -- removing `equals`/`hashCode`
from that fixture makes it **fail**, so it is now a check that Spark honours
the contract rather than a check that the deleted helper worked. A new sibling
pins the documented cost: wit [...]
`build/sbt 'catalyst/testOnly *TransformExpressionSuite'` -- 3 pass.
`build/sbt 'sql/testOnly *KeyGroupedPartitioningSuite
*ProjectedOrderingAndPartitioningSuite *MergeSubplansSuite
*DSv2PlanMergingSuite *PlanMergingSuite'` -- 230 pass. `dev/lint-scala` clean.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 5)
Closes #58002 from peter-toth/SPARK-58769-transform-expression-equality.
Authored-by: Peter Toth <[email protected]>
Signed-off-by: Peter Toth <[email protected]>
(cherry picked from commit 11ea3f7e8bb95d950ba449232666dbc4e74886a7)
Signed-off-by: Peter Toth <[email protected]>
---
.../connector/catalog/functions/BoundFunction.java | 42 ++++++++++
.../expressions/TransformExpressionSuite.scala | 93 ++++++++++++++++++++++
.../sql/execution/planmerging/PlanMerger.scala | 50 +++---------
.../execution/planmerging/MergeSubplansSuite.scala | 88 ++++++++++++++++----
4 files changed, 218 insertions(+), 55 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/BoundFunction.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/BoundFunction.java
index 53a1beb9c4b1..2c8c33f1653f 100644
---
a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/BoundFunction.java
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/BoundFunction.java
@@ -90,6 +90,14 @@ public interface BoundFunction extends Function {
* functions in other catalogs. For example, many catalogs may define a
"bucket" function with a
* different implementation. Adding context, like
"com.mycompany.bucket(string)", is recommended
* to avoid unintentional collisions.
+ * <p>
+ * Two functions that partition data differently must not return the same
name; Spark may
+ * otherwise treat unrelated data as co-partitioned. An override should
return a stable name
+ * across {@code bind} calls, since Spark may bind a function multiple times
and has only this
+ * name to relate two bound instances by. Equal functions must share the
same canonical name;
+ * the reverse is not true, as this name is deliberately coarser and says
nothing about the
+ * rest of a function's state. For whether two transform expressions are the
same expression,
+ * see {@link #equals(Object)}.
*
* @return a canonical name for this function
*/
@@ -100,4 +108,38 @@ public interface BoundFunction extends Function {
// bugs if not replaced before release.
return UUID.randomUUID().toString();
}
+
+ /**
+ * Implementations SHOULD override {@link Object#equals(Object)} and {@link
Object#hashCode()}.
+ * <p>
+ * Spark may bind a function multiple times, so the same transform can be
represented by two
+ * different bound instances. Without a semantic {@code equals} it cannot
tell they are the
+ * same, and misses optimizations such as:
+ * <ul>
+ * <li>keeping a union's keyed partitioning</li>
+ * <li>retaining a reported ordering that matches the partitioning</li>
+ * <li>reusing identical bucketed scans</li>
+ * <li>recognizing two identical subplans</li>
+ * </ul>
+ * Missed matches cost performance only, never correctness.
+ * <p>
+ * Compare whatever state affects behaviour, and keep it stable across
{@code bind} calls.
+ * {@link #canonicalName()} alone is not always enough: two functions can
share a name and still
+ * differ in, say, {@link #resultType()} or a {@link ReducibleFunction}'s
reducers. Two functions
+ * that can produce different values must not compare equal -- the same
comparison decides whether
+ * two ordinary calls to a {@link ScalarFunction} or an {@link
AggregateFunction} are the same
+ * expression, so Spark may otherwise evaluate one where the query asked for
the other.
+ * {@code hashCode} must agree with {@code equals}.
+ * <p>
+ * If this method is overridden, {@link #canonicalName()} should be
overridden as well, so that
+ * equal functions share the same name.
+ */
+ @Override
+ boolean equals(Object other);
+
+ /**
+ * Must agree with {@link #equals(Object)}. See that method.
+ */
+ @Override
+ int hashCode();
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/TransformExpressionSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/TransformExpressionSuite.scala
new file mode 100644
index 000000000000..1c41331eb900
--- /dev/null
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/TransformExpressionSuite.scala
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.expressions
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.connector.catalog.functions.{BoundFunction,
ScalarFunction}
+import org.apache.spark.sql.types.{DataType, IntegerType}
+
+class TransformExpressionSuite extends SparkFunSuite {
+
+ /**
+ * A bound function with a stable canonical name and no `equals` of its own.
A plain class, NOT a
+ * case class: a case class would bring structural `equals`, which is the
connector-provided
+ * comparison these tests are trying to do without.
+ */
+ private class NamedFunction(canonical: String) extends ScalarFunction[Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = canonical
+ override def canonicalName(): String = canonical
+ }
+
+ /** Honours the contract in `BoundFunction#equals`. */
+ private class ComparableFunction extends NamedFunction("test.comparable") {
+ override def equals(other: Any): Boolean =
other.isInstanceOf[ComparableFunction]
+ override def hashCode(): Int = canonicalName().hashCode
+ }
+
+ private val a = AttributeReference("a", IntegerType)()
+ private val b = AttributeReference("b", IntegerType)()
+
+ private def bucket(function: BoundFunction, child: Expression, numBuckets:
Int = 4) =
+ TransformExpression(function, Seq(child), Some(numBuckets))
+
+ test("SPARK-58769: expression equality follows the function's own equals") {
+ // Spark does not derive transform identity itself -- it defers to the
connector, because only
+ // the connector knows which of its state matters. A function that does
not implement `equals`
+ // therefore yields expressions that do not compare equal across separate
binds, which costs
+ // deduplication and reuse but never correctness. See BoundFunction#equals.
+ assert(
+ bucket(new NamedFunction("test.bucket"), a) != bucket(new
NamedFunction("test.bucket"), a),
+ "no equals on the function means no equality across binds")
+
+ val shared = new NamedFunction("test.bucket")
+ assert(bucket(shared, a) == bucket(shared, a), "a shared instance is equal
either way")
+
+ // A function that does implement it gets the deduplication.
+ val left = bucket(new ComparableFunction, a)
+ val right = bucket(new ComparableFunction, a)
+ assert(left.function ne right.function, "the fixture must bind a fresh
instance per call")
+ assert(left == right)
+ assert(left.semanticEquals(right))
+ assert(ExpressionSet(Seq(left, right)).size == 1)
+ }
+
+ test("SPARK-58769: the two comparisons agree for a function that follows the
contract") {
+ // `equals` is the finer comparison and the canonical name the coarser
one, so a function that
+ // overrides both answers both consistently. They still differ in what
they take into account:
+ // `isSameFunction` ignores the arguments, because a join compares
bucket(4, left.id) against
+ // bucket(4, right.id) and recovers the positions separately, while
equality does not.
+ val left = bucket(new ComparableFunction, a)
+ val right = bucket(new ComparableFunction, b)
+ assert(left.function ne right.function, "the fixture must bind a fresh
instance per call")
+ assert(left.function == right.function)
+ assert(left.function.canonicalName() == right.function.canonicalName())
+ assert(left.isSameFunction(right), "the same partition function, arguments
aside")
+ assert(left != right, "but not the same expression, since the arguments
differ")
+ }
+
+ test("SPARK-58769: the function's equals does not override the arguments") {
+ // A coarse comparison on the connector's side does not have to carry the
whole identity: the
+ // transform's arguments and bucket count are compared separately, by
Spark.
+ assert(bucket(new ComparableFunction, a) != bucket(new ComparableFunction,
b),
+ "different argument")
+ assert(bucket(new ComparableFunction, a, 4) != bucket(new
ComparableFunction, a, 8),
+ "different bucket count")
+ }
+}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
index b614985bed9e..0c967421283f 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/planmerging/PlanMerger.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql.execution.planmerging
import scala.collection.mutable
-import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
AttributeMap, AttributeSet, Expression, ExpressionSet, If, Literal,
NamedExpression, Or, SortOrder, TransformExpression}
+import org.apache.spark.sql.catalyst.expressions.{Alias, And, Attribute,
AttributeMap, AttributeSet, Expression, ExpressionSet, If, Literal,
NamedExpression, Or, SortOrder}
import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
import org.apache.spark.sql.catalyst.plans.{Cross, Inner, JoinType, LeftAnti,
LeftOuter, LeftSemi, RightOuter}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join,
LogicalPlan, Project}
@@ -858,49 +858,17 @@ class PlanMerger(
}
}
- // Compares two reported partitioning expression lists. A
TransformExpression must be compared by
- // transform semantics -- `isSameFunction`, i.e. the connector's
`canonicalName` and bucket count
- // -- and NOT by the BoundFunction instance it holds. V2ExpressionUtils
binds the function afresh
- // on every derivation, and while `canonicalName` is a documented obligation
(its default returns
- // a random UUID precisely to force an override), `equals` on BoundFunction
is not required
- // anywhere. So for a connector that meets only the documented contract,
`canonicalized ==` calls
- // two identical `bucket(4, id)` reports different and declines the merge.
Connectors that do
- // define `equals` semantically are unaffected -- Iceberg's
`BaseScalarFunction` compares on
- // `canonicalName` -- but that is a courtesy we should not depend on.
`isSameFunction` is the same
- // notion a storage-partitioned join uses to compare its two sides, but only
that part of it is
- // borrowed: `KeyedShuffleSpec.isExpressionCompatible` matches any two
leaves (it recovers the
- // positions separately), ignores transform children and honours
- // `v2BucketingAllowCompatibleTransforms`, none of which is sound here --
preserving a report
- // means reproducing it exactly, so leaves and children must match too.
- private def sameReportedExpressions(a: Seq[Expression], b: Seq[Expression]):
Boolean = {
- a.length == b.length && a.zip(b).forall {
- case (l: TransformExpression, r: TransformExpression) =>
- // Recurse, so a nested transform's children are compared by transform
semantics too,
- // rather than falling back to the instance-sensitive equality this
method exists to avoid.
- l.isSameFunction(r) && sameReportedExpressions(l.children, r.children)
- case (l, r) => l semanticEquals r
- }
- }
-
- // [[SortOrder.orderingSatisfies]] with the sort keys compared by
`sameReportedExpressions` (see
- // above) rather than semanticEquals. `sameOrderExpressions` needs no
handling here: a reported
- // ordering comes from V2ExpressionUtils.toCatalystOrdering, which always
leaves it empty.
- private def reportedOrderingSatisfies(o1: Seq[SortOrder], o2:
Seq[SortOrder]): Boolean = {
- o2.length <= o1.length && o2.zip(o1).forall { case (required, provided) =>
- provided.direction == required.direction && provided.nullOrdering ==
required.nullOrdering &&
- sameReportedExpressions(Seq(provided.child), Seq(required.child))
- }
- }
-
// The key-grouped partitioning the merged scan must reproduce to keep both
inputs not-worse: the
// two must be equal, so a differing non-empty pair is INCOMPATIBLE (None);
an empty side imposes
// no constraint. Compared in cp's relation space (np's report was remapped
into it by the
- // caller).
+ // caller). A report carrying a `TransformExpression` compares equal across
the two inputs only if
+ // the connector's `BoundFunction` implements `equals` -- Spark does not
derive that identity
+ // itself, see `BoundFunction#equals` -- so a connector that does not gives
up this merge.
private def combineRequiredKeyGroupedPartitioning(
a: Seq[Expression], b: Seq[Expression]): Option[Seq[Expression]] = {
if (a.isEmpty) Some(b)
else if (b.isEmpty) Some(a)
- else if (sameReportedExpressions(a, b)) Some(a)
+ else if (a.map(_.canonicalized) == b.map(_.canonicalized)) Some(a)
else None
}
@@ -909,8 +877,8 @@ class PlanMerger(
// neither satisfies the other they are INCOMPATIBLE (None). An empty
ordering never constrains.
private def combineRequiredOrdering(
a: Seq[SortOrder], b: Seq[SortOrder]): Option[Seq[SortOrder]] = {
- if (reportedOrderingSatisfies(a, b)) Some(a)
- else if (reportedOrderingSatisfies(b, a)) Some(b)
+ if (SortOrder.orderingSatisfies(a, b)) Some(a)
+ else if (SortOrder.orderingSatisfies(b, a)) Some(b)
else None
}
@@ -936,10 +904,10 @@ class PlanMerger(
val kgpDegraded = !dsv2AllowKeyGroupedPartitioningDegradation &&
requiredKeyGroupedPartitioning.nonEmpty &&
!merged.keyGroupedPartitioning.exists(
- sameReportedExpressions(_, requiredKeyGroupedPartitioning))
+ _.map(_.canonicalized) ==
requiredKeyGroupedPartitioning.map(_.canonicalized))
val orderingDegraded = !dsv2AllowOrderingDegradation &&
requiredOrdering.nonEmpty &&
- !reportedOrderingSatisfies(merged.ordering.getOrElse(Nil),
requiredOrdering)
+ !SortOrder.orderingSatisfies(merged.ordering.getOrElse(Nil),
requiredOrdering)
kgpDegraded || orderingDegraded
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
index bf689c73476f..670366ee87ed 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/planmerging/MergeSubplansSuite.scala
@@ -2036,10 +2036,12 @@ class MergeSubplansSuite extends PlanTest {
* exactly as in production, where each scan relation is annotated by its
own derivation.
*/
private def v2ScanReportingOn(
- table: TestV2Table, cols: Seq[String]): DataSourceV2ScanRelation = {
+ table: TestV2Table,
+ cols: Seq[String],
+ funCatalog: FunctionCatalog = TestFreshBindFunctionCatalog):
DataSourceV2ScanRelation = {
val fullOutput = toAttributes(table.schema())
val relation = DataSourceV2Relation(
- table, fullOutput, Some(TestFreshBindFunctionCatalog), None,
CaseInsensitiveStringMap.empty())
+ table, fullOutput, Some(funCatalog), None,
CaseInsensitiveStringMap.empty())
val output = cols.map(c => fullOutput.find(_.name == c).get)
val builder = table.newScanBuilder(CaseInsensitiveStringMap.empty())
builder.asInstanceOf[SupportsPushDownRequiredColumns].pruneColumns(
@@ -2779,11 +2781,11 @@ class MergeSubplansSuite extends PlanTest {
test("SPARK-58549: merge DSv2 scans reporting the same bucket transform
partitioning") {
// Both inputs report `bucket(4, a)`, each derived independently, so each
holds its OWN
// BoundFunction instance -- what production does, since V2ExpressionUtils
binds the function
- // afresh per derivation and a BoundFunction defines no `equals`.
Comparing the two reports by
- // canonical equality would therefore call them different and decline the
merge; they must be
- // compared by transform semantics (isSameFunction), and then the merge
proceeds and the rebuilt
- // scan re-derives the same report. This is the only test that reaches the
transform arm of
- // sameReportedExpressions: an identity-partitioned source reports a plain
attribute instead.
+ // afresh per derivation. The two reports compare equal only because
`TestBucketFunction`
+ // implements `equals`/`hashCode` the way `BoundFunction` asks a connector
to, so this is the
+ // end-to-end check that Spark honours that contract: the merge proceeds
and the rebuilt scan
+ // re-derives the same report. It is also the only test here whose report
is a transform rather
+ // than a plain attribute, which is what an identity-partitioned source
would report.
val q = testRelation.select(
ScalarSubquery(v2ScanReportingOn(v2TableBucketedOnA, Seq("a", "b"))
.groupBy()(sum($"b").as("sum_b"))),
@@ -2800,6 +2802,23 @@ class MergeSubplansSuite extends PlanTest {
s"the preserved partitioning should be on a; got $kgp")
}
+ test("SPARK-58769: decline the merge when the reported transform is not
comparable") {
+ // The mirror of the test above, and the cost this documents: same query,
same reported
+ // `bucket(4, a)`, but the connector's function does not implement
`equals`, so the two
+ // independently bound instances do not compare equal and Spark cannot
tell the two reports
+ // apart from two different partitionings. It does not derive that
identity itself (see
+ // `BoundFunction#equals`), so the merge is declined rather than done on a
report it cannot
+ // verify, and the plan is left alone.
+ val q = testRelation.select(
+ ScalarSubquery(
+ v2ScanReportingOn(v2TableBucketedOnA, Seq("a", "b"),
TestNotComparableFunctionCatalog)
+ .groupBy()(sum($"b").as("sum_b"))),
+ ScalarSubquery(
+ v2ScanReportingOn(v2TableBucketedOnA, Seq("a", "c"),
TestNotComparableFunctionCatalog)
+ .groupBy()(sum($"c").as("sum_c"))))
+ comparePlans(Optimize.execute(q.analyze), q.analyze)
+ }
+
test("SPARK-58549: enforce the required report on the deferred under-Filter
scan build") {
// The above tests fuse scans directly under an Aggregate (no Filter), so
they exercise the
// scan build at the leaf. When the scans sit under an (identical) Filter
the build is instead
@@ -3161,8 +3180,8 @@ private case class TestV2ReportingScan(
* A `FunctionCatalog` that resolves `bucket`, binding it to a FRESH function
instance every time --
* as a real connector does, since `V2ExpressionUtils.loadV2FunctionOpt` binds
the function afresh
* for every derivation of a reported transform. Spark's own
`UnboundBucketFunction` hands back a
- * singleton, which would hide the fact that two independently derived reports
have to be
- * compared by transform semantics rather than by function instance.
+ * singleton, which would hide the fact that relating two independently
derived reports rests on the
+ * function's own `equals` (see `BoundFunction#equals`).
*/
private object TestFreshBindFunctionCatalog extends FunctionCatalog {
override def initialize(name: String, options: CaseInsensitiveStringMap):
Unit = {}
@@ -3183,15 +3202,56 @@ private object TestUnboundBucketFunction extends
UnboundFunction {
}
/**
- * A bound `bucket` with no `equals`, so two instances are never `==` -- which
is what makes the
- * fixture able to catch an instance-sensitive comparison. `canonicalName` is
stable, as the
- * `BoundFunction` contract requires (its default returns a fresh random
UUID); that stable name is
- * what identifies two separately bound instances as the same transform, for a
storage-partitioned
- * join and for the merge's not-worse check alike.
+ * A bound `bucket` that implements `equals`/`hashCode` over the state
identifying it, as
+ * `BoundFunction` asks a connector to. That is what lets Spark recognize two
separately bound
+ * instances as the same transform: `V2ExpressionUtils` binds afresh on every
derivation, so two
+ * reported partitionings hold two different instances and nothing but this
comparison relates them.
+ * `canonicalName` is stable too (its default returns a fresh random UUID),
which is what a
+ * storage-partitioned join compares.
*/
private class TestBucketFunction extends ScalarFunction[Int] {
override def inputTypes(): Array[DataType] = Array(IntegerType, IntegerType)
override def resultType(): DataType = IntegerType
override def name(): String = "bucket"
override def canonicalName(): String = "testcat.bucket"
+
+ override def equals(other: Any): Boolean = other match {
+ case that: TestBucketFunction =>
+ canonicalName() == that.canonicalName() && resultType() ==
that.resultType() &&
+ // `Array` equality is reference identity, so compare the elements.
+ inputTypes().sameElements(that.inputTypes())
+ case _ => false
+ }
+
+ override def hashCode(): Int = canonicalName().hashCode
+}
+
+/**
+ * The same catalog with a bound `bucket` that does NOT implement
`equals`/`hashCode` -- a connector
+ * that meets only the `canonicalName` obligation. Two of its separately bound
instances never
+ * compare equal, so Spark cannot relate two reports of one transform.
+ */
+private object TestNotComparableFunctionCatalog extends FunctionCatalog {
+ override def initialize(name: String, options: CaseInsensitiveStringMap):
Unit = {}
+ override def name(): String = "test_not_comparable"
+ override def listFunctions(namespace: Array[String]): Array[Identifier] =
+ Array(Identifier.of(Array.empty, "bucket"))
+ override def loadFunction(ident: Identifier): UnboundFunction = ident.name()
match {
+ case "bucket" => TestUnboundNotComparableBucketFunction
+ case other => throw new UnsupportedOperationException(s"no such function:
$other")
+ }
+}
+
+private object TestUnboundNotComparableBucketFunction extends UnboundFunction {
+ override def bind(inputType: StructType): BoundFunction = new
TestNotComparableBucketFunction
+ override def description(): String = name()
+ override def name(): String = "bucket"
+}
+
+/** A plain class, so it inherits the identity comparison from `Object`. */
+private class TestNotComparableBucketFunction extends ScalarFunction[Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType, IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "bucket"
+ override def canonicalName(): String = "testcat.bucket"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
