peter-toth commented on code in PR #58002:
URL: https://github.com/apache/spark/pull/58002#discussion_r3811322570
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/BoundFunction.java:
##########
@@ -100,4 +135,49 @@ default String canonicalName() {
// bugs if not replaced before release.
return UUID.randomUUID().toString();
}
+
+ /**
+ * Implementations SHOULD override {@link Object#equals(Object)} and {@link
Object#hashCode()}.
+ * <p>
+ * Spark binds a function afresh every time it converts a partitioning or
ordering reported by a
+ * source, so two reports of the same transform -- from two scans of the
same table, say -- hold
+ * two bound instances. Spark compares the partition transform expressions
built from them with
+ * ordinary expression equality, which reaches this method. With the
inherited identity comparison
+ * the two are never equal, and Spark cannot deduplicate the expressions,
reuse a scan that
+ * reports them, or recognize two subplans as the same. That costs
optimizations only: results
+ * are unaffected either way.
+ * <p>
+ * Compare whatever state affects behaviour. {@link #canonicalName()} on its
own is not
+ * necessarily enough: it answers a narrower question and is allowed to be
coarse, so two
+ * functions can share one and still differ in {@link #inputTypes()}, {@link
#resultType()},
+ * {@link #isResultNullable()}, {@link #isDeterministic()}, or a {@link
ReducibleFunction}'s
+ * reducers. Comparing the name IS sufficient for an implementation whose
names already encode
+ * everything that distinguishes its functions -- one name per argument
type, say. Only the
+ * implementation knows which of its state matters, which is why Spark
cannot derive this
+ * comparison itself. Note that a coarse name does not have to carry the
whole identity: Spark
+ * compares the transform's arguments separately.
+ * <p>
+ * This is the FINER of the two comparisons, so two functions that compare
equal must also return
+ * the same {@link #canonicalName()}; an implementation overriding this
method should override
+ * that one too, rather than leave it at its default.
+ * <p>
+ * Two functions that can produce DIFFERENT values must NOT compare equal.
Being too coarse is the
+ * direction that costs more than optimizations, and it is not confined to
partition transforms: a
+ * {@link ScalarFunction} or an {@link AggregateFunction} is held by the
expression Spark builds
+ * for an ordinary call to it, so the same {@code equals} decides whether
two such calls are the
+ * same expression, and Spark may then evaluate one where the query asked
for the other.
+ * <p>
+ * Whatever is compared must be stable across {@code bind} calls, and {@code
hashCode} must agree
+ * with {@code equals}, since Spark puts these expressions in hash-based
collections.
+ */
+ @Override
+ boolean equals(Object other);
Review Comment:
Thanks, that reads much better -- taken almost as written.
I kept two things from the old text. The first sentence, on Spark binding
afresh, because without it the requirement looks arbitrary: a connector author
whose bound function is effectively a singleton has no reason to expect
`equals` to be called on two instances at all. And the clause about
`ScalarFunction`/`AggregateFunction` calls, because that's the only place the
doc says a too-coarse `equals` can change results rather than just cost
performance.
Everything else went, including the enumeration of behaviour-affecting state
and the "why Spark can't derive this itself" paragraph -- that one answers a
Spark reviewer's question, not a connector author's, so it belongs in the PR
description.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/BoundFunction.java:
##########
@@ -90,6 +90,41 @@ default boolean isDeterministic() {
* 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. They are
+ * indistinguishable to Spark, which can then treat unrelated data as
co-partitioned and produce
+ * a join with wrong results, so include whatever tells them apart --
argument types, for
+ * instance -- in the name. This is the only requirement here that guards
results rather than
+ * performance.
+ * <p>
+ * An overriding implementation should keep the name stable across calls,
and across separate
+ * {@code bind} calls for the same function: Spark binds a function afresh
every time it converts
+ * a partitioning or ordering reported by a source, so when the two
instances are not the same
+ * object it has only this name to relate them by when deciding whether two
sides of a join are
+ * co-partitioned, and so whether a storage-partitioned join can avoid a
shuffle. An unstable name
+ * is not unsafe -- it just makes overriding pointless, since it relates
nothing to anything.
+ * <p>
+ * This name answers only that question -- whether two transforms are the
same PARTITION FUNCTION,
+ * ignoring their arguments, which a join needs because it compares {@code
bucket(4, left.id)}
+ * against {@code bucket(4, right.id)}. It is deliberately not a complete
identity for the bound
+ * function: it says nothing about {@link #inputTypes()}, {@link
#resultType()},
+ * {@link #isResultNullable()}, {@link #isDeterministic()}, a {@link
ReducibleFunction}'s
+ * reducers, or any other state the implementation carries. For the separate
question of whether
+ * two partition transform expressions are the same expression, see {@link
#equals(Object)}.
+ * <p>
+ * The two questions are related in one direction: this name is the COARSER
of the two, so two
+ * functions that compare equal must return the same canonical name, while
the same name does not
+ * make them equal. An implementation that overrides {@link #equals(Object)}
should therefore
+ * override this method as well -- keeping the default while claiming two
instances are equal says
+ * they are the same expression but not the same partition function, which
costs it exactly the
+ * co-partitioning the name exists to enable.
+ * <p>
+ * Leaving the default in place is allowed: it opts the function out of
being recognized as
+ * equivalent to anything, which costs optimizations and nothing else. Two
instances are never
+ * equivalent, so Spark forgoes co-partitioning between them; and because
the default is not even
+ * stable across two calls on ONE instance, such a function is not the same
partition function as
+ * itself, so a partitioning it takes part in can cost a shuffle that a
stable name would have
+ * avoided. Overriding it is the way out of all of that.
Review Comment:
Taken, thanks. I also dropped the paragraph below it about what leaving the
default costs -- the default method's own comment already says a function is
never equivalent to another, even itself, so it wasn't earning its length.
--
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]
