(spark) branch branch-4.x updated: [SPARK-58769][SQL][FOLLOWUP] Clarify the BoundFunction equality contract

[sz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

szehon-ho 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 a1e8b7b9ba0f [SPARK-58769][SQL][FOLLOWUP] Clarify the BoundFunction equality contract
a1e8b7b9ba0f is described below

commit a1e8b7b9ba0f1b546a67be2af00daa90690fcb20
Author: Szehon Ho <sz...@gmail.com>
AuthorDate: Fri Aug 28 13:41:39 2026 -0700

    [SPARK-58769][SQL][FOLLOWUP] Clarify the BoundFunction equality contract
    
    ### What changes were proposed in this pull request?
    
    Clarify the `BoundFunction.equals` documentation by:
    
    - adding valid aggregate-query rejection to the existing list of consequences when separately bound functions do not compare equal;
    - adding concise connector examples for stateless functions and functions with bound state; and
    - correcting nearby test comments that implied `UnboundFunction.bind` must return a fresh instance.
    
    ### Why are the changes needed?
    
    The current documentation says missed equality matches affect performance only. However, if repeated scalar function calls bind to distinct objects that do not compare equal, a query that selects and groups by the same call may fail with `MISSING_AGGREGATION`.
    
    Also, `UnboundFunction.bind` permits both shared and newly created bound-function instances.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    Documentation and test-comment changes only. `git diff --check` passed.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: OpenAI Codex
    
    Closes #58363 from szehon-ho/codex/fix-boundfunction-javadoc.
    
    Authored-by: Szehon Ho <sz...@gmail.com>
    Signed-off-by: Szehon Ho <sz...@gmail.com>
    (cherry picked from commit 024120dc960517021a887b51865220f544340362)
    Signed-off-by: Szehon Ho <sz...@gmail.com>
---
 .../connector/catalog/functions/BoundFunction.java | 17 ++++++++++---
 .../expressions/TransformExpressionSuite.scala     | 10 ++++----
 .../execution/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 2c8c33f1653f..a5b729a6fa33 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 @@ public interface BoundFunction extends Function {
   /**
    * 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:
+   * 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:
    * <ul>
+   *   <li>accepting a valid query that selects and groups by the same scalar function call</li>
    *   <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
@@ -133,6 +133,15 @@ public interface BoundFunction extends Function {
    * <p>
    * If this method is overridden, {@link #canonicalName()} should be overridden as well, so that
    * equal functions share the same name.
+   *
+   * <h4>Connector implementation examples</h4>
+   * <ul>
+   *   <li>A stateless {@code StringLengthFunction}, which returns a string's length, can compare
+   *   by concrete class.</li>
+   *   <li>A {@code TruncateFunction} that stores a configured width must include the width.</li>
+   *   <li>A {@code ToTimestampFunction} that stores a configured time zone must include the time
+   *   zone.</li>
+   * </ul>
    */
   @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 1c41331eb900..3cee9cb13001 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 670366ee87ed..cd16ac4ea665 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)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org