From c438fd4e2fcee9c58df629413fc09b04e3a9a5b0 Mon Sep 17 00:00:00 2001 From: Szehon Ho Date: Thu, 27 Aug 2026 14:32:14 -0700 Subject: [PATCH] [SPARK-58769][SQL][FOLLOWUP] Clarify the BoundFunction equality contract --- .../catalog/functions/BoundFunction.java | 17 ++++++++--- .../TransformExpressionSuite.scala | 10 +++---- .../planmerging/MergeSubplansSuite.scala | 29 ++++++++----------- 3 files changed, 30 insertions(+), 26 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 2c8c33f1653fa..a5b729a6fa33b 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 @@ -112,16 +112,16 @@ default String canonicalName() { /** * Implementations SHOULD override {@link Object#equals(Object)} and {@link Object#hashCode()}. *

- * 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: + * Spark may bind a function multiple times, so the same function call can be represented by two + * different bound instances. Without a semantic {@code equals} it cannot tell they are the same. + * This can prevent Spark from: *

- * Missed matches cost performance only, never correctness. *

* 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 @@ -133,6 +133,15 @@ default String canonicalName() { *

* If this method is overridden, {@link #canonicalName()} should be overridden as well, so that * equal functions share the same name. + * + *

Connector implementation examples

+ * */ @Override boolean equals(Object other); 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 index 1c41331eb9009..3cee9cb130011 100644 --- 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 @@ -49,9 +49,9 @@ class TransformExpressionSuite extends SparkFunSuite { 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. + // the connector knows which of its state matters. Separate instances of a function that does + // not implement `equals` therefore yield expressions that do not compare equal. 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") @@ -62,7 +62,7 @@ class TransformExpressionSuite extends SparkFunSuite { // 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.function ne right.function, "the fixture must use distinct function instances") assert(left == right) assert(left.semanticEquals(right)) assert(ExpressionSet(Seq(left, right)).size == 1) @@ -75,7 +75,7 @@ class TransformExpressionSuite extends SparkFunSuite { // 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 ne right.function, "the fixture must use distinct function instances") assert(left.function == right.function) assert(left.function.canonicalName() == right.function.canonicalName()) assert(left.isSameFunction(right), "the same partition function, arguments aside") 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 670366ee87ed2..cd16ac4ea6657 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 @@ -2779,13 +2779,12 @@ 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. 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. + // Both inputs report `bucket(4, a)` and derive the report independently. This fixture returns a + // distinct BoundFunction from each bind. The reports compare equal because `TestBucketFunction` + // implements `equals`/`hashCode` as `BoundFunction` recommends, so 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"))), @@ -3177,11 +3176,9 @@ 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 relating two independently derived reports rests on the - * function's own `equals` (see `BoundFunction#equals`). + * A `FunctionCatalog` whose `bucket` function returns a new bound instance for every bind. + * `V2ExpressionUtils` binds each independently derived report, and this fixture exercises the case + * where those calls do not share object identity. */ private object TestFreshBindFunctionCatalog extends FunctionCatalog { override def initialize(name: String, options: CaseInsensitiveStringMap): Unit = {} @@ -3203,11 +3200,9 @@ private object TestUnboundBucketFunction extends UnboundFunction { /** * 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. + * `BoundFunction` asks a connector to. This fixture returns a distinct instance for each bind, so + * two independently derived reports are related only by this comparison. `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)