[SPARK-50593][SQL] SPJ: Support truncate transform via generalized ReducibleFunction API

[metanil]

What changes were proposed in this pull request?

This PR adds Storage Partitioned Join (SPJ) support for the truncate partition transform. The approach generalizes the ReducibleFunction API to accept arbitrary transform parameters, so SPJ can reason about any parameterized transform (bucket, truncate, future ones) through one code path.

Key changes:

• Generalized reducer ReducibleFunction.reducer(Literal<?>[], ReducibleFunction, Literal<?>[]). Parameters are V2 Literals, so a connector sees each value together with its DataType. Spark tries this overload first and falls back to the deprecated reducer(int, ..., int) only when it is not implemented, both sides are the same function, and each side has a single integer parameter, so existing connectors (e.g., Iceberg, which implements only the old API) continue to work unchanged.
• TransformExpression refactor: literal parameters (e.g., bucket numBuckets, truncate width) now live inside children rather than a bespoke numBucketsOpt: Option[Int] field, and V2ExpressionUtils resolves bucket through the same generic path as every other transform.
• Transform identity carries the parameters. TransformFunctionId records each argument's shape in order (literal, nested transform, or column), and both isSameFunction and hasSameReducedKeys compare it, so truncate(col, 3) and truncate(col, 5) are different transforms and reduce onto different key spaces.
• Rewrites preserve parameters. KeyedShuffleSpec.createPartitioning retargets a transform through a helper that leaves literal parameters untouched, and alias projection never substitutes a literal, for partitionings and orderings alike. So e.g. SELECT data AS d, 2 AS w over a truncate(data, 2) partitioning keeps SPJ, and its ordering still matches the partitioning.

Why are the changes needed?

Today a join on tables partitioned by truncate(col, N) always shuffles, even when both sides share identical partitioning. The write-side was fixed by SPARK-40295 (Allow v2 functions with literal args in write distribution and ordering), but the read/join side was never enabled.

Previous work in #49211 (@szehon-ho) explored direct support for transforms with literal arguments by adjusting the SPJ paths to recognize them. This PR generalizes the reducer API so the compatibility check is function-agnostic.

It builds on two changes now on master. SPARK-59121 records when a join has reduced both sides onto a common key space (TransformExpression.reducedWith) and compares those key spaces by transform identity; SPARK-59688 (#58943) stops the join pairing raw keys that still need reducing, the issue raised earlier in this review. Both compare transforms by identity, which is why identity has to carry the parameters once non-bucket transforms can reduce.

Does this PR introduce any user-facing change?

Yes, for connector/catalog authors:

• New overload ReducibleFunction.reducer(Literal<?>[], ReducibleFunction, Literal<?>[]) (@since 4.4.0); reducer(int, ReducibleFunction, int) is deprecated.
• No action required for existing connectors; they keep working via the fallback. A connector that implements only the old API is covered by a dedicated LegacyBucketFunction test fixture.

For end users, queries joining tables partitioned by compatible truncate transforms (identical widths, or reducible pairs like truncate(3) and truncate(5)) now avoid shuffle via SPJ.

How was this patch tested?

43 new tests across KeyGroupedPartitioningSuite, TransformExpressionSuite and ProjectedOrderingAndPartitioningSuite: truncate SPJ end to end, transform identity (parameter values and positions, nested transforms), parameters surviving alias projection, and the reducer dispatch and backward-compatibility paths.

cc @szehon-ho @aokolnychyi @sunchao @peter-toth

Was this patch authored or co-authored using generative AI tooling?

Yes.

Generated-by: Claude Code (Opus 5)

[sunchao]

[P1] This widens support from transforms with one direct reference child to any transform whose collectLeaves() returns one leaf. Because Transform arguments can themselves be nested Transforms, V2ExpressionUtils can materialize shapes like outer(years(k)) and outer(days(k)). The new gate accepts both, keyPositions later maps them only by leaf k, and TransformExpression.isSameFunction compares only the outer function name plus literals. That can make storage partitionings with different nested child semantics look compatible and let SPJ skip a required shuffle, which can drop join matches. Keep the old direct-reference constraint, or compare the full non-literal child semantics before admitting these transforms.

[ 🤖 posted by Codex on behalf of sunchao 🤖 ]

[metanil]

Thanks @sunchao for the review. Good point. I didn't think about nested transform. This is indeed widening the transforms, as collectLeaves() go through all the children in nested expressions.
Like you mentioned, adding direct-reference constraints here would be enough. I'll make that change by getting non literal children, and use same existing check (size and reference). With that we also don't have to then worry about TransformExpression.isSameFunction comparing only outer function.

With regards to adding support to nested transform for SPJ, very interesting idea, but we can take that as separate PR.

[sunchao]

[P1] extractParameters forwards raw Catalyst Literal.value objects into ReducibleParameters. For string literals, Spark stores UTF8String, while the new public API documents string parameters and exposes getString() as a java.lang.String cast. A connector implementing the documented string-parameter case will get ClassCastException from getString(0) or be forced to depend on Spark internals. Convert literal values to connector-facing external values by dataType before constructing ReducibleParameters.

[ 🤖 posted by Codex on behalf of sunchao 🤖 ]

[metanil]

I agree, due UTF8String, we'll get ClassCastException. I think same case for Decimal type.
I'll make the change to handle these, and possibly a little more generic way for future proofing.

[sunchao]

[P2] The new generalized reducer API accepts ReducibleParameters on both sides, and this file even models zero-literal transforms as ReducibleParameters([]). But this dispatch only invokes the generalized overload when both sides are non-empty. Mixed cases such as parameterized-vs-zero-parameter transforms instead fall back to reducer(otherFunction), so connectors that correctly implement the new generalized overload for those cases are never invoked; the default legacy path can even throw UnsupportedOperationException. If mixed arity is meant to be unsupported, the new API/docs should say that explicitly. Otherwise this should dispatch through the generalized overload whenever either side wants that path.

[ 🤖 posted by Codex on behalf of sunchao 🤖 ]

[metanil]

@sunchao Good point. I was trying to fallback to existing behavior, but hadn't considered parameterized-vs-zero-parameter case, which it should follow the new generalized path. With regards making mixed arity "unsupported", currently I can't think of an valid example, but I also can't think of not to support it either. So, I suppose, we can let the implementor reducer decide to have it or not.

I will make the change that will dispatch through the generalized overloaded reducer().

[szehon-ho]

i think it makes more sense to have it the other way. the old one should call the new (more generic one). we can have the temporary code on spark side:

val thisParams = extractParameters(thisExpr)
val otherParams = extractParameters(otherExpr)
if (singleIntParam(thisParams) && singleIntParam(otherParams)) {
  Option(thisFunction.reducer(thisParams.getInt(0), otherFunction, otherParams.getInt(0)))
    .orElse(Option(thisFunction.reducer(thisParams, otherFunction, otherParams)))
} else if (!thisParams.isEmpty && !otherParams.isEmpty) {
  Option(thisFunction.reducer(thisParams, otherFunction, otherParams))
} else {
  Option(thisFunction.reducer(otherFunction))
}

imo, its not much uglier than this code :)

In practice, Iceberg is the only implementation of this so we can hopefully remove this ugly wrapper from spark side.

[metanil]

@szehon-ho Yes, this make sense. My initial idea was to not call "deprecated" method, and when time to remove it would be only from 1 single file (ReducibleFunction), but the approach is not natural (or norm), where deprecated code calls the newer method.
When there's time to remove the method, we can remove from both, would not be a big deal, and can't be missed too.
I'll make this change along with @sunchao mixed arity case.

[szehon-ho]

version is not right

[metanil]

@szehon-ho oh good eye! I worked on this last year :) Will update to reflect current version.

[metanil]

@sunchao @szehon-ho Addressed all comments, PTAL

Additionally, I'm now handling all "null" or unimplemented exception (UOE) from reducer() and displaying log warnings.

cc @peter-toth

[peter-toth]

Summary

Adds SPJ support for the truncate partition transform by generalizing the reducer API: literal parameters now live inline in TransformExpression.children, are surfaced to connectors through a new ReducibleParameters container, and a generalized ReducibleFunction.reducer(ReducibleParameters, ReducibleFunction, ReducibleParameters) overload replaces the bucket-only reducer(int, …, int) signature.

Prior state and problem. Bucket's numBuckets was carried as a sidecar numBucketsOpt: Option[Int] field on TransformExpression, outside the expression tree. The SPJ reducer dispatch in TransformExpression.reducer only knew how to fan out two cases — (Some, Some) → reducer(int, …, int) or anything else → reducer(otherFunction) — so any parameterized transform other than bucket (e.g. truncate(col, N)) was structurally invisible to the reducer dispatch and always shuffled. The write side was fixed in SPARK-40295; the read/join side never was. This PR supersedes the earlier exploration in #49211 by generalizing the API rather than special-casing transforms one at a time.

Design approach. Drop the numBucketsOpt sidecar and let literals participate in children. Then expose those literals to connectors as a typed parameter container, route the SPJ reducer dispatch through a single generalized overload, and keep the deprecated int-API alive (with default throw UOE) so existing connectors like Iceberg 1.10.0 continue to work. The dispatch tries the deprecated signature first for the single-int case (Iceberg shape), falls back to the new overload on UnsupportedOperationException, and goes straight to the new overload for everything else.

Key design decisions.

• Literals are now inline Expression children, not a typed field. TransformExpression.collectLeaves() is overridden to strip them so the existing partitioning.scala/EnsureRequirements consumers still see exactly one leaf attribute per transform.
• ReducibleParameters is a brand-new public API rather than reusing V2Literal/Literal; it exposes typed getters (getInt, getLong, getString, getDouble, getFloat, get) but deliberately hides the source DataType.
• Spark's own BucketFunction was switched to override only the new API. Backward compat for connectors that override only the deprecated method is covered by a dedicated LegacyBucketFunction test fixture.
• extractParameters converts UTF8String → java.lang.String and Decimal → java.math.BigDecimal, but otherwise passes Catalyst-internal values straight through.

Implementation sketch. New public API in sql/catalyst/.../functions/ (ReducibleParameters.java, generalized ReducibleFunction.reducer overload). Catalyst plumbing in TransformExpression.scala (new literalChildren, extractParameters, tryReduce, dispatch logic; overrides for collectLeaves and canonicalized). V2-to-Catalyst resolution simplified in V2ExpressionUtils (the BucketTransform case is dropped; bucket now resolves through the generic NamedTransform path). Physical layer in partitioning.scala updates KeyedPartitioning.supportsExpressions to count non-literal children and KeyedShuffleSpec.createPartitioning to preserve literals when rewriting children. DistributionAndOrderingUtils.resolveTransformExpression simplified. Test fixtures in transformFunctions.scala add TruncateFunction.reducer via the new API, plus IntegerTruncateFunction and LegacyBucketFunction. New tests in KeyGroupedPartitioningSuite.

Behavioral changes worth calling out.

• truncate(col, N) now plans KeyedPartitioning where it previously planned UnknownPartitioning. The existing test non-clustered distribution: V2 function with multiple args was renamed to clustered distribution: … to reflect this.
• EnsureRequirementsSuite exprA–exprD flipped from Literal(1..4) to AttributeReferences because the new code distinguishes literals from refs in transform children. Tests using bucket(N, exprA) now exercise a [Literal(N), AttributeReference] shape rather than [Literal(N), Literal(1)]. Worth a sanity check that no existing assertion relied on the prior Literal shape.

General

• @sunchao (Codex, May 15) and @szehon-ho (May 18) already raised the substantive issues with the earlier commit. The author has addressed: the collectLeaves() widening on KeyedPartitioning.supportsExpressions (now uses children.filterNot(_.isInstanceOf[Literal]) directly), the mixed-arity dispatch (now routes through the generalized overload), the dispatch ordering preference (deprecated-first, new fallback per szehon's suggestion), and the @since version on ReducibleParameters. Nice turnaround on all of those.
• The fix for @sunchao's extractParameters P1 is only partial. The current code converts UTF8String → String and Decimal → BigDecimal, but the original critique was structural: convert by dataType rather than special-casing each type. As written, BinaryType (byte[]), CalendarIntervalType, ArrayType/MapType Catalyst values, etc. still flow through the catch-all value.asInstanceOf[AnyRef] and a connector implementing the documented "any type" claim on ReducibleParameters will hit Catalyst-internal classes. The principled fix is to drive the conversion off the literal's DataType (e.g., via existing CatalystTypeConverters.convertToScala) — or, taken further, to surface V2 literals directly through the API; see inline #6.
• The PR description claim "New overload ReducibleFunction.reducer(ReducibleParameters, ...) with a default that delegates to the deprecated single-int signature for backward compatibility" is out of date with the code. The new default throws UnsupportedOperationException; it's TransformExpression.reducer's Spark-side dispatch that tries the deprecated signature first. Worth updating the description so reviewers don't get a wrong mental model of the fallback mechanism.
• The collectLeaves override on TransformExpression (inline #3 below) is the kind of contract divergence that's better fixed at the abstraction layer it crosses. I've put up #56088 as a follow-up that migrates the SPJ call sites in partitioning.scala and EnsureRequirements.scala off _.collectLeaves() — to Expression.references where set semantics suffices, and a new Expression.collectAttributes(): Seq[Attribute] where positional Seq is required. Pure refactor against today's master. Once #56088 lands, this PR can rebase on top and drop the override here entirely; the migrated call sites already filter literals correctly.

Suggested improvements

1. ReducibleParameters is missing getBigDecimal. extractParameters converts Decimal → java.math.BigDecimal on the wire, but the container exposes no decimal getter, so connectors with a decimal parameter must use the untyped get(index) and cast manually — exactly the type-safety the wrapper was added to prevent. (Inline #6 below proposes a structural alternative that obviates this gap entirely.) [sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleParameters.java:114]
2. canonicalized override is dead code and the doc justification doesn't hold. SPJ uses isSameFunction, not canonicalized/semanticEquals, so the stated motivation ("bucket(4, tableA.id) and bucket(4, tableB.id) semantically equal") is moot — and AttributeReference.canonicalized keeps exprId, so the override wouldn't deliver that equality even if SPJ did read it. The override otherwise duplicates the default path. Remove. [sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:93]
3. collectLeaves override breaks the universal TreeNode contract — drop on rebase after #56088. The follow-up migrates the SPJ call sites off _.collectLeaves() so the override here can simply be removed; see the General section. [sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:111]
4. UnboundTruncateFunction.bind() claims BinaryType and LongType support, but neither branch actually works. BinaryType → TruncateFunction mismatches (StringType, IntegerType) inputTypes() and a getUTF8String read; LongType → IntegerTruncateFunction mismatches (IntegerType, IntegerType) inputTypes() and a getInt read. Plus the IntegerTruncateFunction doc claim about "incompatible partition structures" is wrong. Drop the unused branches, or split per type with coverage. [sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/functions/transformFunctions.scala:293]
5. isSameFunction compares literals by value but ignores positional structure. The check tolerates literals at different positions in children and ignores nested-transform children entirely (compares only the outer function name + literal values), so bucket(4, days(col)) and bucket(4, hours(col)) would be reported as the same function. A structural per-position compare closes both gaps. [sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:75]
6. Structural alternative — reuse V2Literal instead of introducing ReducibleParameters. V2Literal is already public and already used everywhere else in the connector API; passing it through the reducer signature preserves dataType() (no Catalyst-internal-types leak), makes inline #1 moot (no typed getters to maintain), and shrinks extractParameters to a Catalyst-Literal → V2-Literal conversion. Worth weighing before locking the new public class in. [sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleParameters.java:40]

[peter-toth]

extractParameters in TransformExpression.scala:171 converts Decimal → java.math.BigDecimal on the way in, but this class exposes no getBigDecimal accessor. A connector with a decimal parameter therefore has to fall back to the untyped Object get(int index) and cast by hand — exactly the type-safety the wrapper was introduced to provide. Either add a getBigDecimal getter alongside the others, or remove the Decimal special-case in extractParameters and document the supported types explicitly. Concretely:

/**
 * Get parameter at index as BigDecimal.
 * @throws ClassCastException if parameter is not a BigDecimal
 * @throws IndexOutOfBoundsException if index is invalid
 */
public java.math.BigDecimal getBigDecimal(int index) {
    return (java.math.BigDecimal) values.get(index);
}

Same gap exists for any other type the Spark side might convert (binary, interval, etc.) — see the ## General note about driving extractParameters off DataType. Inline #6 proposes a structural alternative that would make this whole class of issue moot.

[peter-toth]

The doc claim that motivates this override doesn't hold, and the override doesn't add anything over the inherited default. Two pieces:

The motivation doesn't hold. The doc says "we need bucket(4, tableA.id) and bucket(4, tableB.id) to be semantically equal so SPJ can be triggered". SPJ doesn't use canonicalized/semanticEquals on TransformExpression though — it dispatches through isSameFunction/isCompatible (see KeyedShuffleSpec.isExpressionCompatible at partitioning.scala:1082). And even if it did, this override wouldn't deliver the claim: AttributeReference.canonicalized (namedExpressions.scala:324) normalizes the name to "none" but keeps the original exprId, so two different attributes with the same name still canonicalize to different forms. bucket(4, tableA.id) and bucket(4, tableB.id) aren't made equal by either the override or the default.

The body is equivalent to the default. Literal is a LeafExpression, so Literal.canonicalized returns the literal unchanged — the per-child case l: Literal => l branch produces the same result as l.canonicalized. The override therefore matches withCanonicalizedChildren exactly, except it skips the Canonicalize.execute post-pass that the default routes through (a no-op for TransformExpression today, but a future change to Canonicalize would silently fail to apply here).

Recommended: remove the override entirely.

[metanil]

Good catch, yes, not needed. Was meant to remove it.

[metanil]

Removed

[peter-toth]

This override breaks the universal TreeNode.collectLeaves contract — the base method (TreeNode.scala:315) returns "every node in the tree where children.isEmpty", but this returns column references only, hiding the literal leaves. Functionally correct for current SPJ call sites (partitioning.scala:492/498/982/1037, EnsureRequirements.scala:89/412/417/780), but a quiet semantic divergence — any future caller of collectLeaves on a TransformExpression will silently get wrong answers.

The cleaner shape is to migrate the SPJ call sites off _.collectLeaves() and use the right helper at each:

• Expression.references where set semantics suffice (existence/size checks, single-attribute lookup).
• A new Expression.collectAttributes(): Seq[Attribute] where a positional Seq is required (RowOrdering.create, reorder, attributes.zip(clustering)) — AttributeSet deduplicates and its ordering is LinkedHashSet-implementation-detail rather than contract.

I've sent #56088 as the follow-up doing exactly that. Pure refactor against today's master (current TransformExpression has no literal children, so the migrated call sites produce identical results). Once it's merged, this override can be dropped on rebase, and 55885's representation change (literals as inline children) lands cleanly through the already-migrated call sites.

[metanil]

Yes, exactly this. I was getting away with the context of TransformExpression as it always meant for Attributes, but with respect to TreeNode.collectLeaves contract, it was indeed breaking.
I was also trying handle by filter Literals during usages, but meant modifying all the callers.

collectAttributes() is exactly what I needed, and will use it once #56088 land on main branch.

Thanks for the PR @peter-toth

[peter-toth]

Actually, now I think we don't even need a new Expression.collectAttributes(), but we can use Expression.references() in SPJ planning. Adjusted the PR.

[peter-toth]

I've merged #56088 so collectLeaves() override is no longer needed.

[metanil]

@peter-toth awesome! thanks.

[peter-toth]

This dispatch claims BinaryType and LongType support, but neither branch actually works:

• StringType | BinaryType => TruncateFunction — TruncateFunction.inputTypes() declares (StringType, IntegerType) and produceResult calls input.getUTF8String(0). A BinaryType row (raw byte[] underneath) mismatches inputTypes() at bind-time, or — if the framework lets it through — fails at runtime on the getUTF8String read.
• IntegerType | LongType => IntegerTruncateFunction — IntegerTruncateFunction.inputTypes() declares (IntegerType, IntegerType) and produceResult calls input.getInt(0). Same kind of mismatch on a LongType row, plus its resultType() is IntegerType so even if you cast input down, the column type is wrong.

None of the phantom branches are exercised by tests in this PR — every truncate test uses StringType and dispatches to TruncateFunction correctly. So the broken branches add API surface that connector authors might trust without coverage to defend it.

Adjacent issue: the doc on IntegerTruncateFunction ("different integer truncate widths produce incompatible partition structures") is wrong: truncate(v, W1) is reducible to truncate(v, W2) whenever W2 % W1 == 0 (snap to a coarser width), and when neither divides the other both reduce to truncate(_, lcm(W1, W2)). Same structural argument as bucket's GCD case.

Three options, in order of cleanliness:

1. Drop the unused branches and IntegerTruncateFunction from this PR — only StringType truncate is part of the SPJ story being delivered.
2. Split per type (IntegerTruncate / LongTruncate / BinaryTruncate) with the right inputTypes()/resultType()/produceResult and add at least one test per branch.
3. Make each existing object polymorphic on the input type (declare wider inputTypes(), dispatch in produceResult) and add tests.

[metanil]

@peter-toth fixed Removed unsupported datatype in the test. Also I converted IntegerTruncateFunction with proper reducible method, along with new unit test for it.

[peter-toth]

This compares literals by value but ignores positional structure of children. Two consequences:

1. Position-agnostic. Two transforms with the same function and the same literal values but different interleavings of literals/non-literals are reported as the same. In practice the V2 function catalog pins down argument order per function name, so this is unlikely to manifest, but the doc claims "same semantics as other" — the check is weaker than that without the catalog assumption.

2. Nested-transform children. V2ExpressionUtils.toCatalyst recurses through toCatalystTransformOpt, so children can include nested TransformExpressions (sunchao's earlier P1 raised this for supportsExpressions). bucket(4, days(col)) and bucket(4, hours(col)) both have outer function bucket and literalChildren = [Literal(4)], so this returns true even though the inner transforms differ. KeyedPartitioning.supportsExpressions rejects nested-transform shapes from SPJ planning, so this doesn't bite SPJ today, but isSameFunction is used elsewhere (isCompatible, suite assertions) where the false positive would leak.

A structural per-position walk closes both gaps and lets literalChildren shrink to just the extractParameters helper:

def isSameFunction(other: TransformExpression): Boolean = {
  function.canonicalName() == other.function.canonicalName() &&
    children.length == other.children.length &&
    children.zip(other.children).forall {
      case (a: Literal, b: Literal) => a == b
      case (a: TransformExpression, b: TransformExpression) => a.isSameFunction(b)
      case (_: Literal, _) | (_, _: Literal) => false
      case (_: TransformExpression, _) | (_, _: TransformExpression) => false
      case _ => true   // both refs/attrs — column identity intentionally ignored
    }
}

[metanil]

@sunchao raised this issue too, and I stuck with the current limitation of SPJ with only 1 Reference.
But I like the idea of generalizing it to support nested cases as well.

I'll try to follow the suggestion and make the change.

Thanks @peter-toth for suggestion.

[metanil]

@peter-toth
Updated isSameFunction method which is recursive:

• outer function name + per-position children;
• recurses into nested transforms;
• ignores column identity
  So, now bucket(4, years(c)) is not the same as bucket(4, days(c)).

[peter-toth]

Structural alternative — reuse V2Literal instead of introducing ReducibleParameters. Surfacing this even though it's late in the cycle, because it's the kind of design choice worth weighing before locking in a new public class.

The proposal: drop ReducibleParameters entirely and have the generalized reducer take org.apache.spark.sql.connector.expressions.Literal (the V2 literal type already used everywhere else in the connector API):

default Reducer<I, O> reducer(
    org.apache.spark.sql.connector.expressions.Literal<?>[] thisParams,
    ReducibleFunction<?, ?> otherFunction,
    org.apache.spark.sql.connector.expressions.Literal<?>[] otherParams) {
  throw new UnsupportedOperationException();
}

Connector use:

@Override
public Reducer<UTF8String, UTF8String> reducer(
    Literal<?>[] thisParams,
    ReducibleFunction<?, ?> otherFunc,
    Literal<?>[] otherParams) {
  if (otherFunc != TruncateFunction) return null;
  int thisWidth = (Integer) thisParams[0].value();
  int otherWidth = (Integer) otherParams[0].value();
  ...
}

What this fixes simultaneously:

1. Inline Removed reference to incubation in README.md. #1's gap evaporates. No typed getters to maintain — connectors call .value() and dispatch on .dataType(). No more "we forgot getBigDecimal" / "we'll need getCalendarInterval" / etc.
2. The extractParameters partial-conversion concern (General-section bullet Removed reference to incubation in Spark user docs. #2) goes away. V2Literal carries dataType() alongside the value, so the connector interprets it correctly regardless of whether it's String, BigDecimal, byte[], CalendarInterval, etc. No Catalyst-internal types leaking, no type-by-type special-casing — extractParameters shrinks to a Catalyst-Literal → V2-Literal conversion (essentially the inverse of what V2ExpressionUtils.toCatalyst already does for V2 literals).
3. One fewer public class to learn / maintain / stabilize. V2Literal is already public, already @Evolving → stable, already used by every connector that authors V2 transforms (Expressions.literal(N), Expressions.bucket(N, col)). Receiving them back through the reducer API is symmetric round-trip with how transforms are constructed — V2 connectors never have to cross the Catalyst boundary in this public surface.

Trade-offs being honest about:

• Slightly more verbose at the connector use site ((Integer) params[0].value() vs params.getInt(0)). Real but small.
• Java generics + arrays awkwardness; Literal<?>[] produces unchecked-warning ceremony. List<Literal<?>> or Literal<?>... varargs are cleaner alternatives.
• The deprecated int-API → new-API dispatch shim still needed (single int → Literal<Integer>).

cc @sunchao @szehon-ho — would value your read on this trade-off, given your existing reviews of ReducibleParameters. Should the API rebase on V2Literal, or stay with the new class as drafted?

[metanil]

@peter-toth @sunchao I've addressed most of the feedbacks except one (see below)

• extractParameters is DataType driven. Catalyst-internal literal values are converted to
  external types via CatalystTypeConverters.convertToScala (e.g. UTF8String → String,
  Decimal → java.math.BigDecimal) instead of per-type special-casing, so connectors never see
  Catalyst-internal classes.

• Correctness for nested transforms (bucket(4, years(ts))), since a transform argument can
  itself be a transform:

  • Updated isSameFunction method which is recursive:
    • outer function name + per-position children;
    • recurses into nested transforms;
    • ignores column identity
      So, now bucket(4, years(c)) is not the same as bucket(4, days(c)).
  • The reducer dispatch is guarded (nonLiteralChildrenSame): a reducer derived from the outer
    literal parameters is only applied when every non-literal child is structurally identical,
    preventing reduction across unrelated nested transforms (e.g. bucket(4, years(ts)) vs
    bucket(2, days(ts))).
  • KeyedShuffleSpec.canCreatePartitioning opts out of nested transforms so the
    repartition-to-match path never flattens an inner transform away.

• Removed two now-redundant overrides on TransformExpression:

  • collectLeaves — SPJ planning reads Expression.references after your SPARK-57038, so the
    literal-filtering override is no longer needed.
  • canonicalized — it only re-implemented the default per-child canonicalization.

• Test fixtures (transformFunctions.scala): TruncateFunction (String) implements the new API;
  IntegerTruncateFunction is reducible via lcm (a third reducer shape distinct from bucket-GCD
  and string-prefix-min); LegacyBucketFunction exercises the deprecated-only connector path.

One item I haven't addressed yet is ReducibleParameters as we are waiting from @sunchao feedback on it. Once confirmed, I'll either add missing typed getters ( cases like BigDecimal etc) there or switch to use Literal<?>[]

[peter-toth]

First, apologies if our earlier comments were unclear -- our suggestion to use references was specifically about consumption sites (see comment #55885 (comment)), not about widening this gate. We had not asked for the SPJ gate to be widened, and the recursive isSameFunction change does not require it.

The new commit bundles three semi-independent changes here and in TransformExpression:

1. Recursive isSameFunction in TransformExpression -- the truncate fix. ✓
2. Widening isSupportedTransform to transform.references.size == 1 -- now admits nested transforms like bucket(4, years(col)), and also bucket(4, cast(col)), bucket(4, col + 1), etc.
3. New hasOnlyReferenceArgs gate on KeyedShuffleSpec.canCreatePartitioning, plus the nonLiteralChildrenSame guard inside TransformExpression.reducer -- both exist only because (2) admits shapes where the prior code was unsound.

(2) is a coverage expansion, not a fix for any existing bug. We'd like this PR to ship the truncate fix without the coverage expansion:

• Restore the previous gate that limited isSupportedTransform to "single non-literal child that is a column reference."
• Drop hasOnlyReferenceArgs -- it exists only to narrow back what the widened gate would otherwise let through.
• Drop the nonLiteralChildrenSame guard inside reducer. It patches a real latent bug (without it, bucket(4, years(c)) vs bucket(2, days(c)) would silently reduce via gcd), but that bug is unreachable once the strict gate is restored: every nested-transform shape that reaches the reducer has already been ruled out by the gate itself.

Keep the recursive isSameFunction machinery -- it's the right shape for the truncate fix (position-agnostic literal compare) and is well-documented; future widening can build on it.

[metanil]

Thanks @peter-toth
Reverted.

Yep I agree, for nested transforms support for SPJ, we can discuss separately.

[peter-toth]

This duplicates KeyedPartitioning.isReference in partitioning.scala -- same shape, same semantics. Conditional on inline #1 above (revert the widening, remove hasOnlyReferenceArgs), this method becomes unused and can be deleted. Otherwise please consolidate -- either call KeyedPartitioning.isReference from here, or hoist the helper into a shared utility used by both sites.

[metanil]

Moved into companion objects for reusability.

[peter-toth]

CatalystTypeConverters.convertToScala returns config-dependent runtime types for DateType and TimestampType:

• DateType → java.time.LocalDate if spark.sql.datetime.java8API.enabled is true, otherwise java.sql.Date.
• TimestampType → java.time.Instant if the same flag is true, otherwise java.sql.Timestamp.

So the runtime class of a date/timestamp value inside ReducibleParameters is a function of the executor's SQLConf rather than of the column's DataType. A connector implementing reducer(ReducibleParameters, ReducibleFunction, ReducibleParameters) cannot just write params.get(0).asInstanceOf[LocalDate] -- it must handle both forms, or risk a ClassCastException on whichever form it didn't anticipate.

This is an additional point in favor of the V2Literal-array alternative discussed at #55885 (comment): V2 Literal<T> already carries the internal representation along with the DataType. Connectors receive a stable, self-describing (value, dataType) pair and decide for themselves how to interpret it -- which is the right contract for a public reducer API.

[metanil]

Sounds good. I have Literal[] version ready go too. I'm doing some testing. Just waiting for @sunchao 's feedback.

[peter-toth]

@szehon-ho , @sunchao WDYT?

[szehon-ho]

Thanks @peter-toth, I'm on board with reusing V2Literal over introducing ReducibleParameters. A few reasons it's the better long-term contract, and how I'd address the two weaknesses you called out:

Why V2Literal:

• It's already the public, @Evolving literal type connectors use to construct transforms (Expressions.bucket(N, col), Expressions.literal(N)), so receiving them back through reducer is a symmetric round-trip rather than a new concept to learn.
• It's self-describing: value() + dataType() lets the connector interpret the value itself. That structurally closes the issues we kept hitting with ReducibleParameters — the missing getBigDecimal, and the config-dependent Date/Timestamp runtime types from convertToScala. With V2Literal carrying the internal representation alongside the DataType, extractParameters collapses to a Catalyst-Literal -> V2-Literal conversion (the inverse of what V2ExpressionUtils.toCatalyst already does), with no per-type special-casing.
• Type-future-proofing: a new DataType needs zero API changes here, vs. ReducibleParameters needing a new typed getter per type. One fewer public class to stabilize and maintain.

On the perceived weaknesses:

1. Complex/nested types leaking Catalyst internals. In practice transform parameters are scalars (numBuckets, truncate width), so this doesn't bite. I'd rather make that explicit than leave it ambiguous: let's document on the reducer API that parameters are scalar literals, and have extractParameters reject non-scalar literal children rather than silently forwarding an ArrayData/MapData/InternalRow. That keeps the contract honest without expanding scope. (Note this caveat would apply to ReducibleParameters too, only worse — it'd need a new getter and still leak.)

2. Array-generics ceremony (Literal<?>[]). Agree it's awkward. Let's use varargs (Literal<?>... thisParams / Literal<?>... otherParams) — it avoids the unchecked-warning noise, reads cleanly at the connector call site, and the int-API -> new-API shim just passes Expressions.literal(numBuckets). List<Literal<?>> would also work but varargs is the lightest touch.

The slightly-more-verbose call site ((Integer) params[0].value() vs params.getInt(0)) is a fair trade — it's the same idiom connectors already use against InternalRow, and Iceberg is migrating off the deprecated int-API regardless, so there's no existing implementation that loses out.

cc @sunchao — feel free to chime in if you want, since you reviewed ReducibleParameters too.

[sunchao]

Will do. Thanks for the ping 🙏

[metanil]

@szehon-ho Thanks for the input, Just a note, I like the idea of varargs, but both Scala and Java doesn't support in the middle of the argument. Seems like we're stuck with Literal<?>[].
But I like the idea of not exposing the internal complex type/scalar literals.

[sunchao]

+1 on using V2 Literal<?>[] too, given that it can:

• Preserves both value() and dataType() instead of storing untyped Objects.
• Avoids maintaining separate getters like getInt(), getString(), and missing future getters.
• Handles typed nulls and new Spark data types without changing the public API.
• Avoids Spark converting values into configuration-dependent Java types.
• Reuses the existing connector literal abstraction instead of introducing another public class.

[peter-toth]

Please drop this comment, the ticket prefixes identify the purpose of the tests.

[sunchao]

Sorry for the delay. Will take another look!

[sunchao]

[P1] Please preserve the column/literal argument layout before invoking a reducer. The strict supportsExpressions gate only requires one direct column reference; it does not require that reference to occupy the same child position on both sides. extractParameters then removes the literal positions, and the latest commit removed the positional nonLiteralChildrenSame guard.

This is reachable for a public V2 function with two same-typed arguments. For example, f(id, 2) and f(4, store_id) both pass the gate. The generalized reducer sees only [2] and [4] and can report compatibility. I reproduced this on the current head with the integer-truncate reducer: both transforms were admitted, isCompatible returned true, and matching join value 3 mapped to reduced key 0 on the left versus key 3 on the right. SPJ can therefore skip a required shuffle and lose the match.

Please restore a positional-shape check before reducer dispatch: literal values may differ, but literal and column-reference slots must align on both sides.

Posted by Codex on behalf of @sunchao.

[sunchao]

[P2] Try(Option(oldReducer)).orElse(genericReducer) only evaluates the generalized overload when the deprecated call throws. A documented null return becomes Success(None), so Try.orElse returns it without executing the fallback.

I reproduced this on the current head with a function implementing both overloads: the deprecated overload returned null, the generalized overload returned a valid reducer, and reducers(...).isDefined was still false. A connector retaining the deprecated method while implementing broader compatibility through the replacement API will therefore miss a valid reducer and add an unnecessary shuffle.

Please inspect the inner Option and invoke the generalized overload after either a failed deprecated call or None. A regression test should cover legacy-null/new-success.

Posted by Codex on behalf of @sunchao.

[peter-toth]

sameArgumentLayout and isSameFunction (L76-86) are now two separate children.zip(...) walkers. The literal axis differing is deliberate — equality for "same transform" vs. "values may differ, a Reducer reconciles them". But the non-literal handling drifted apart, and sameArgumentLayout ended up strictly weaker:

position pair	isSameFunction	sameArgumentLayout
(literal, literal)	l1 == l2	true — intended
(transform, transform)	recursive isSameFunction	case _ => true
(col, transform)	isColumnRef && isColumnRef	case _ => true

So sameArgumentLayout treats any two non-literals as interchangeable, and its correctness rests entirely on the supportsExpressions gate guaranteeing the single non-literal child is always a plain column ref. On its own it would let two structurally different transforms that share a function name and literal layout reduce (e.g. bucket(4, days(c)) vs bucket(4, hours(c))).

That's fine as the code stands. But since these were a single parameterized helper a commit ago (childrenMatch), it'd be nice to bring sameArgumentLayout back in sync with isSameFunction so the precondition is correct on its own rather than gate-dependent (future-proof):

/** Per-position match; `literalsMatch` is the only axis that varies between callers. */
private def childrenMatch(other: TransformExpression)
    (literalsMatch: (Literal, Literal) => Boolean): Boolean =
  children.length == other.children.length &&
    children.zip(other.children).forall {
      case (l1: Literal, l2: Literal) => literalsMatch(l1, l2)
      case (t1: TransformExpression, t2: TransformExpression) => t1.isSameFunction(t2)
      case (c1, c2) => TransformExpression.isColumnRef(c1) && TransformExpression.isColumnRef(c2)
    }

def isSameFunction(other: TransformExpression): Boolean =
  function.canonicalName() == other.function.canonicalName() &&
    childrenMatch(other)(_ == _)

// reducer precondition: same layout/structure, literal *values* may differ
private def sameArgumentLayout(other: TransformExpression): Boolean =
  childrenMatch(other)((_, _) => true)

Equally fine to leave as-is if you'd rather not touch it now — mainly flagging so the gate-dependency is a conscious choice.

[metanil]

@peter-toth I like the idea. Yea, very similar to my parameterized helper created for nested transform. Will make the change.
Thanks.

[peter-toth]

Version stamp: this should be 4.3.0, not 5.0.0. The companion SPARK-57038 merged to both master (5.0.0) and branch-4.x (4.3.0); assuming this PR is likewise backported, it first ships in 4.3.0, so per the "version of the branch it first ships in" rule both this @since and the @Deprecated(since = "5.0.0") on the deprecated overload below (L118) should read 4.3.0. If this is intended master-only, ignore.

Suggested change

	* @since 5.0.0
	* @since 4.3.0

[peter-toth]

Thanks @metanil — all my review comments are addressed, LGTM. I plan to merge this Wednesday if there are no further asks or concerns from @szehon-ho or @sunchao.

[sunchao]

[P1] Reducer-compatible transforms must not take the raw partition-key equality fast path.

This widened gate now admits parameterized transforms such as xor(x, 1). With spark.sql.sources.v2.bucketing.pushPartValues.enabled=true, spark.sql.sources.v2.bucketing.allowCompatibleTransforms.enabled=true, and partially clustered distribution disabled, consider two tables containing x IN (0, 1) partitioned by identity(x) and xor(x, 1). Both scans advertise sorted raw partition keys [0, 1], but partition 0 contains x=0 on the identity side and x=1 on the xor side (partition 1 is reversed). KeyedShuffleSpec.isCompatibleWith treats the reducer-compatible expressions plus equal unreduced keys as directly compatible, so EnsureRequirements skips reducer reconciliation and GroupPartitionsExec; a shuffle-free equality join silently returns zero rows instead of two.

Please restrict the raw-key-equality fast path to identical transform semantics, or force compatible-but-different transforms through reduced-key reconciliation. Add end-to-end SPJ regressions for identity-vs-parameterized and parameterized-vs-parameterized transforms whose raw key arrays coincide but whose physical partitions differ.

[peter-toth]

@metanil this has gone a month without a rebase, and something landed in the meantime that bears directly on it.

SPARK-58769 (#58002, merged to master and branch-4.x) dropped the name-based transform comparison, so TransformExpression now compares its bound functions with equals, and it documents that contract on BoundFunction. A connector whose bind returns instances that do not compare equal no longer gets a match by function name. That matters here, because the truncate reducers are looked up per side and compared. There is also an open follow-up, #58363, clarifying the same contract.

Could you rebase on master and adjust for it? I will take another pass once it is green.

[metanil]

@peter-toth Thanks for the update. Will update from master adjust the change.

I was addressing some naive solution for @sunchao 's P1 feedback, but had created other issues. I'll probably revisiting it a week after next.

[peter-toth]

@metanil a heads-up on sunchao's P1 about the raw partition-key equality fast path: it is not specific to this PR.

The same hole is live on master, through the identity-versus-transform arm SPARK-56182 added. A table partitioned by identity(id), joined with one partitioned by a transform that reorders its key space, where both sides report the key list [0, 1]: the join returns 0 rows instead of 2, with no shuffle and no GroupPartitionsExec in the plan. I measured that on master and filed SPARK-59688, with the fix in #58943. KeyedShuffleSpec.isCompatibleWith there asks areKeysCompatible with allowReduce = false, so a pair whose keys still need reducing onto one key space is no longer read as lined up, and the join takes the push branch that reduces and regroups.

So there is nothing for you to do here about that P1. Once #58943 lands it falls away on your rebase, and the gate widening in this PR inherits the fix.

[metanil]

@peter-toth Thanks 🙏 . I'll rebase it, and push the necessary change.

[metanil]

Thanks @peter-toth for SPARK-59121 and SPARK-59688 — between them they fix the co-partitioning issues this PR originally tried to address (reduced key spaces are now marked and compared via TransformExpression.reducedWith, and raw keys that still need reducing are no longer paired as they stand), so I've dropped that half and merged master into this branch.

Btw, @peter-toth I modified the your TransformFunctionId to records each argument's shape as well.

Thanks @sunchao for the original review. The raw-key issue you raised is now in the master with peters fix.

[peter-toth]

Re-checked from scratch through 18f6f8f. Nothing I raised in the earlier rounds regressed, and the one open P1 on the PR, r3679486000, is closed by SPARK-59688 on master: isCompatibleWith asks areKeysCompatible(allowReduce = false), so the identity-vs-transform arm no longer reads a pair whose keys still need reducing as lined up.

The rebase moved a lot, so one line of re-orientation. The co-partitioning half is gone, TransformFunctionId now records each argument's shape in order, and isSameFunction and hasSameReducedKeys are both derived from that one value. That is the right shape: a sidecar field cannot express where a parameter sits, and truncate's width sits at argument 1 while bucket's count sits at argument 0, so moving the parameters into children is what makes the generalization possible at all. Findings 1 and 4 are both "one step too specific" rather than "wrong direction", and I have them as one measured patch.

The findings below are all first raised here.

Blocking

• 1. The literal parameter survives the partitioning projection but not the ordering one (late catch): dc0afc98b66 fixed outputPartitioning and left outputOrdering on the shared projectExpression, so ProjectExec(id AS pk, 32 AS w) reports the ordering bucket(w#2, pk#1) over a partitioning of bucket(32, pk#1). Measured, along with the one-clause fix; the two are otherwise out of sync, which is exactly what GroupPartitionsExec.outputOrdering's coalescing branch filters on. inline: sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:97
• 2. No end-to-end test for the identical-width truncate join (late catch): both new end-to-end truncate tests set V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS, which is off by default, so they only cover the reducer path. The case the description leads with is the one the default config takes, and it is covered only by a single-table checkQueryPlan. inline: sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:6295

Non-blocking

• 3. A typed-null literal parameter still reaches the generalized overload (late catch): the null clause here keeps it out of the deprecated overload only. Measured: bucket(null, id).reducers(bucket(4, id)) returns Some(BucketReducer(4)), because BucketFunction reads the null as 0 and gcd(0, 4) is 4. inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:315
• 4. isSameFunction is not reflexive (late catch): identity is re-asking a question supportsExpressions already owns, so a transform with a non-column-ref argument is not the same function as itself, while ShuffleSpec.isCompatibleWith documents reflexivity as assumed. Dropping the Option fixes it and takes 17 lines with it; measured. inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:116
• 5. The deprecated-fallback guard compares name(), not canonicalName() (new): name() has no uniqueness contract, and canonicalName() is both the documented one and what functionId already uses. The two disagree in both directions. inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:336
• 6. IntegerTruncateFunction shares its canonical name with TruncateFunction (late catch): the two partition data differently, which BoundFunction.canonicalName's javadoc forbids, so truncate(strCol, 3) and truncate(intCol, 3) are the same transform. inline: sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/functions/transformFunctions.scala:494

Alternatives

• 7. Detect an unimplemented overload rather than infer it from a caught UnsupportedOperationException (late catch): an implemented reducer that throws UnsupportedOperationException from inside is read as unimplemented and silently falls through to the deprecated overload. Optional, with real counter-arguments. inline: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:322

Minor

• 8. Four comments still describe a dispatch order and a guard shape that changed (new): one of them contradicts the assertion in the test that uses its fixture. inline: sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/functions/transformFunctions.scala:670
• 9. The two ReducibleFunction defaults can recurse into each other (late catch): the natural migration shim delegates back, and the resulting StackOverflowError is fatal, so Try does not catch it. inline: sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleFunction.java:158
• 10. Two gaps in the description (new): the fallback rule omits the condition 22cd902c0d4 added, that both sides must be the same function. And "Paths that rewrite a transform's children (KeyedShuffleSpec.createPartitioning, alias projection) go through one helper" does not hold for the ordering alias projection, see finding 1.

[peter-toth]

Finding 1. dc0afc98b66 closed this for outputPartitioning and left outputOrdering on the shared projectExpression. DataSourceV2ScanExecBase.outputOrdering derives the ordering from the same k.expressions (v2BucketingPartitionKeyOrderingEnabled is on by default), and AliasAwareQueryOutputOrdering.outputOrdering projects it through projectExpression directly, so the literal is still substituted there.

Measured on this head, with a dummy leaf reporting KP([bucket(32, id)]) and outputOrdering = [bucket(32, id) ASC], under ProjectExec(id AS pk, 32 AS w):

partitioning:  keyedpartitioning(transformexpression(bucket, 32, pk#1), ...)
ordering:      List(transformexpression(bucket, w#2, pk#1) ASC NULLS FIRST)
ordering refs: {w#2, pk#1}
ExpressionSet(partitioning.expressions).contains(ordering.head.child): false

Control, same plan without 32 AS w: ordering transformexpression(bucket, 32, pk#3), and contains is true.

That last line is the consequence. GroupPartitionsExec.outputOrdering's coalescing branch keeps exactly the sort orders whose expression is a partition key expression:

val keyExprs = ExpressionSet(keyedPartitionings.flatMap(_.expressions))
child.outputOrdering.filter(order => keyExprs.contains(order.child))

and its comment states the premise this breaks: "The child's outputOrdering should already be in sync with the partitioning". With the literal substituted the filter drops the key-derived ordering, so a downstream sort that would have been eliminated is kept. A GroupPartitionsExec over a ProjectExec over a BatchScanExec is the ordinary shape of a join leg, so this is the default-config path. The claim itself stays value-correct, since w is the constant 32 in every row, which is why nothing fails. It just stops matching.

The fix I would suggest is central rather than per-site, at sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/AliasAwareOutputExpression.scala:74:

      case e: Expression if !e.isInstanceOf[Literal] && aliasMap.contains(e.canonicalized) =>
        aliasMap(e.canonicalized).toSeq ++ (if (e.containsChild.nonEmpty) Seq(e) else Seq.empty)

A Literal then stays put for every consumer, which is what you want everywhere: an alias over a constant is not another name for the same value, it is a column that happens to hold it. projectPartitionExpression can then go away entirely, 18 lines lighter.

I ran it, together with finding 4's change, on this head. Green: TransformExpressionSuite, ShuffleSpecSuite, KeyGroupedPartitioningSuite, ProjectedOrderingAndPartitioningSuite, GroupPartitionsExecSuite, EnsureRequirementsSuite, PlannerSuite (479 tests), plus ReplaceIntegerLiteralsWithOrdinalsDataframeSuite and RemoveRedundantSortsSuite. No test in sql/core partitions or orders by a bare literal, so nothing depended on the old behaviour.

Either way the new ProjectedOrderingAndPartitioningSuite test wants the ordering half asserted next to the partitioning one, since that is the pair that has to stay in sync:

    val orderedChild = DummyLeafExecWithPartitioningAndOrdering(
      output = Seq(id),
      partitioning = KeyedPartitioning(Seq(bucketExpr), keys1d),
      ordering = Seq(SortOrder(bucketExpr, Ascending)))
    val orderedProject = ProjectExec(Seq(pk, w), orderedChild)
    val projectedOrdering = orderedProject.outputOrdering
    assert(projectedOrdering.head.child.asInstanceOf[TransformExpression].children.head
      === Literal(32),
      "the literal parameter must survive the ordering projection too")
    val projectedKp = orderedProject.outputPartitioning.asInstanceOf[KeyedPartitioning]
    assert(ExpressionSet(projectedKp.expressions).contains(projectedOrdering.head.child),
      "the projected ordering must stay in sync with the projected partitioning")

If you prefer to keep the change narrow instead, give AliasAwareQueryOutputOrdering the same TransformExpression case projectPartitionExpression has.

[peter-toth]

Finding 2. Both new end-to-end truncate tests set V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS, which is off by default, so between them they only cover the reducer path. The case the description leads with, identical widths matched by isSameFunction with no reducer at all, is covered only by the single-table checkQueryPlan at line 549. That is the path the default config takes and the one most users will hit.

It works. The test is one copy of this one without the withSQLConf. Measured on this head: 0 shuffles and the right rows, with allowCompatibleTransforms left at false.

  test("SPARK-50593: identical truncate widths trigger SPJ under the default config") {
    val table1 = "trunc_same_a"
    val table2 = "trunc_same_b"
    val partitions = Array(
      Expressions.apply("truncate", Expressions.column("data"), Expressions.literal(3)))

    createTable(table1, columns, partitions)
    sql(s"INSERT INTO testcat.ns.$table1 VALUES " +
      "(0, 'apple', CAST('2022-01-01' AS timestamp)), " +
      "(1, 'grape', CAST('2021-01-01' AS timestamp)), " +
      "(2, 'orange', CAST('2020-01-01' AS timestamp))")

    createTable(table2, columns, partitions)
    sql(s"INSERT INTO testcat.ns.$table2 VALUES " +
      "(10, 'apple', CAST('2022-01-01' AS timestamp)), " +
      "(20, 'grape', CAST('2021-01-01' AS timestamp)), " +
      "(30, 'orange', CAST('2020-01-01' AS timestamp))")

    // No withSQLConf: identical widths are the same transform, so this has to work on the
    // default config, where allowCompatibleTransforms is off.
    val df = sql(
      s"""
         |${selectWithMergeJoinHint(table1, table2)}
         |$table1.id AS left_id, $table2.id AS right_id
         |FROM testcat.ns.$table1 JOIN testcat.ns.$table2
         |ON $table1.data = $table2.data
         |ORDER BY $table1.id
         |""".stripMargin)

    assert(collectShuffles(df.queryExecution.executedPlan).isEmpty,
      "identical truncate widths are the same function, so no shuffle is needed")
    checkAnswer(df, Seq(Row(0, 10), Row(1, 20), Row(2, 30)))
  }

It cannot pass on base: supportsExpressions required children.size == 1 there, which truncate(data, 3) fails, and the pre-PR version of the test at line 549 asserted UnknownPartitioning(0) for exactly that reason.

[peter-toth]

Finding 3. This clause keeps a typed null out of the deprecated overload, which is what r3503138262 asked for. The generalized overload still receives it: literalParamsMatchInputTypes compares only dataType, and noComplexLiteralParams looks only at dataType too, so Literal(null, IntegerType) passes both gates.

All three bundled reducers then read the value with .value().asInstanceOf[Int], and in Scala null.asInstanceOf[Int] is 0. Measured on this head:

bucket(null, id).isSameFunction(bucket(4, id)):  false
bucket(null, id).literalParamsMatchInputTypes:   true
bucket(null, id).reducers(bucket(4, id)):        Some(BucketReducer(4))
bucket(null, id).isCompatible(bucket(4, id)):    true

BucketFunction.reducer read the null as 0, computed gcd(0, 4) = 4, and returned a reducer, so Spark declares the pair reducible and reduces the null side by % 4. That is the same fabricated zero as the deprecated path, one overload up.

A null parameter belongs with the other things the gate refuses, rather than with something the javadoc warns about. Then isSingleInt needs no null clause of its own:

    if (!sameArgumentLayout(other) ||
        !literalParamsMatchInputTypes || !other.literalParamsMatchInputTypes ||
        !noComplexLiteralParams || !other.noComplexLiteralParams ||
        literalChildren.exists(_.value == null) ||
        other.literalChildren.exists(_.value == null)) {
      return None
    }

If you would rather let a null through, the javadoc needs to say so, because right now it tells an implementor to read value() as "Spark's internal representation" and only warns about array lengths.

[peter-toth]

Finding 4. This arm gives up the identity when an argument is not a literal, a nested transform or a plain column ref, so functionId is None and isSameFunction answers false for such a transform against itself. ShuffleSpec.isCompatibleWith documents the opposite at sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:1586: "Note that Spark assumes this to be reflexive, symmetric and transitive". The new TransformExpressionSuite test pins the non-reflexivity ("never the same, not even as itself").

It is reachable under v2BucketingShuffleEnabled. KeyedShuffleSpec.createPartitioning builds the shuffled side's expression from the other side's clustering key, so a join ON t1.id = t2.b + 1 over a bucket(4, id) table yields bucket(4, b + 1) on the shuffled side, paired with bucket(4, id) on the keyed one. keyPositions reconciles them, one reference each, and the two are co-located by construction, since createPartitioning built the expression for exactly that. Master's isSameFunction said true for that pair; this head says false.

The root cause is that identity is being asked two questions. "Which transform is this" is identity's; "may SPJ reason about these argument shapes" is KeyedPartitioning.supportsExpressions', which already asks it and is the load-bearing gate on the scan. Once identity stops re-asking it, the whole thing gets shorter:

  lazy val functionId: TransformFunctionId = {
    import TransformFunctionId.ArgumentShape
    TransformFunctionId(function.canonicalName(), children.map {
      case l: Literal => ArgumentShape.Param(l)
      case t: TransformExpression => ArgumentShape.Nested(t.functionId)
      case _ => ArgumentShape.Slot
    })
  }

with Column renamed to Slot, since it now covers any non-parameter argument. What that discriminates is unchanged where it matters:

• truncate(id, 2) and truncate(2, store_id) stay apart, the Param positions differ
• bucket(4, years(c)) and bucket(4, days(c)) stay apart, Nested is still matched above
• bucket(32, c) and bucket(16, c) stay apart

and the knock-on simplifications fall out:

• isSameFunction becomes functionId == other.functionId
• reducedKeySpace goes back from the for-comprehension to reducedWith.map(partner => Set(functionId, partner))
• reducedTogetherWith loses its SparkException.internalError arm, which its own comment already called unreachable, so it is a one-liner again
• the SparkException import goes

Net -17 lines in TransformExpression.scala.

On whether the wider Slot is safe. Identity is column-blind by construction, and the safety comes from keyPositions plus supportsExpressions, not from identity. For two non-column-slot transforms to be compared at all, both have to be createPartitioning products, since the scan gate refuses such a shape, and those are built from one spec onto one key set, so "same function" is the right answer there. I could not construct a case where the wider slot co-locates rows that do not belong together; if you see one, that settles it the other way.

I ran this on this head. Green: TransformExpressionSuite, ShuffleSpecSuite (41), then KeyGroupedPartitioningSuite, ProjectedOrderingAndPartitioningSuite, GroupPartitionsExecSuite, EnsureRequirementsSuite, PlannerSuite (438). Four assertions had to flip, and three of them are the ones that enshrine the non-reflexivity:

• TransformExpressionSuite, "a transform with a non-reference argument has no identity" -- becomes "a non-parameter argument is a slot, whatever its shape"
• the plusOne assertion in "transform identity is per-argument-position"
• the add assertion in KeyGroupedPartitioningSuite's "isSameFunction is a structural per-position walk"
• the shapes.forall(_.functionId.isDefined) fixture guard in "isSameFunction and hasSameReducedKeys agree on every pair", which can instead gain a non-reference shape to the fixture list, since the agreement now holds over those too

Pushed both changes on top of 18f6f8f in case cherry-picking is easier than redoing them: SPARK-50593-simplify-identity-and-alias-projection on my fork. Two independent commits, so you can take either on its own:

• 4c51b96 -- this finding, the total functionId plus the four test assertions that flip
• e93ae9b -- finding 1, the one-clause projectExpression rule, projectPartitionExpression deleted, and the ordering assertion added to ProjectedOrderingAndPartitioningSuite

75 insertions and 78 deletions over 6 files. Same content I ran the suites against; take it, adapt it or ignore it.

[peter-toth]

Please hold off on this one, and on 4c51b96, for now. I found a case where the total functionId looks wrong, and I'm still looking into it. I'll follow up here.

[peter-toth]

Following up on the hold above. I was wrong here: a transform over b + 1 must never be the same as one over a column, and your None is what guarantees that on this head.

The total functionId makes bucket(4, b + 1) the same as bucket(4, x), and that loses rows:

• KeyedShuffleSpec.createPartitioning shuffles a side onto bucket(4, b + 1) when its join key is b + 1.
• A later join that clusters that side on the bare b pairs it with a bucket(4, x) table, because keyPositions maps a partition expression to a cluster key through its reference.

With 4c51b96 on top of this head, t1 JOIN plain p ON t1.id = p.b + 1 JOIN t3 ON p.b = t3.x returns 0 of 7 rows. This head returns all 7.

Master has the same bug, since its isSameFunction ignores the arguments. I opened SPARK-59887 and #59165 for it. It makes isSameFunction and reducers refuse a transform whose argument is not a bare Attribute, keeps such a layout from being the one other children are shuffled onto, and adds tests for the shapes it fixes. Could you rebase onto it once it lands? Three things to watch in the conflict:

• The new argumentsAreAttributes skips literal children, so it carries over to your model as it is.
• It also refuses a struct field argument, which your Option treats as a Column. A side shuffled onto bucket(4, s.a) pairs with one on bucket(4, s.b) in a join on s, and loses rows the same way. Please keep that refusal, either through the new guard or by making ArgumentShape.Column a bare Attribute only.
• The first new test also asserts two shuffles. This head gives the right rows but plans three, since it has no canCreatePartitioning clause yet, so it passes only once that clause is in.

Please don't take 4c51b96. e93ae9b (finding 1) is unrelated and still stands.

[metanil]

Thanks @peter-toth for the update. I was exploring both findings (1 and 4) yesterday. I'll for wait for your PR #59165 and rebase it here. I'll try to preemptively pull it before the merge to see how it fits here, and be ready.

[peter-toth]

Finding 5. name() has no uniqueness contract. Function.name() is a display name, and Spark's own fixtures return it both bare (BucketFunction, "bucket") and catalog-qualified (ShuffleSpecSuite.FakeBucket, "test.fakeBucket"). canonicalName() is the one documented for this question, and it is what functionId already compares two hundred lines up.

The two disagree in both directions.

• Over-admits. Two different bound functions may share a name(). This PR's own UnboundTruncateFunction binds TruncateFunction or IntegerTruncateFunction by column type, and both return "truncate", so a pair of those passes this clause.
• Under-admits. BoundFunction.canonicalName's javadoc exists for the case where one implementation is loaded under two catalog names: "loaded using different names, like test.func_name and prod.func_name". A connector whose name() carries the catalog then fails this clause for two equivalent functions, and loses the fallback the commit was protecting.

Suggested change

	function.name() == other.function.name() =>
	function.canonicalName() == other.function.canonicalName() =>

[peter-toth]

Finding 6. TruncateFunction above returns the same "truncate" here, and the two partition data differently: one takes a string prefix, the other snaps an int down to a grid. BoundFunction.canonicalName's javadoc rules that out: "Two functions that partition data differently must not return the same name; Spark may otherwise treat unrelated data as co-partitioned."

isSameFunction keys on the canonical name plus the argument shapes, so truncate(strCol, 3) and truncate(intCol, 3) are the same transform today. Nothing in the suite joins those two tables, so it is latent. But these fixtures are what a connector author reads as the reference implementation of the API this PR is generalizing, and the contract they break is the one SPJ rests on.

Suggested change

	override def canonicalName(): String = name()
	override def canonicalName(): String = "truncate(int)"

and "truncate(string)" on TruncateFunction. Leave both name()s as "truncate", so the function catalog still resolves them the same way.

[peter-toth]

Finding 7. Optional, and a design point rather than a defect.

The dispatch reads "the connector did not implement this overload" off a caught UnsupportedOperationException, and the new javadoc now states that convention. But it is not a property of the overload, it is a property of one call. An implemented generalized reducer that throws UnsupportedOperationException from inside, say a connector operation it cannot serve for these parameters or a require in a helper, is read as unimplemented. The pair then falls through to the deprecated overload, which may hand back a reducer for parameters the generalized one refused. Silent, and the "implements no reducer" warning that would otherwise appear is suppressed by the very fallback that fired.

Whether the method is overridden is answerable directly, and there is precedent right next door: V2ExpressionUtils.findMethod already does this for ScalarFunction's magic method. Sketch:

  // Memoized per BoundFunction class, since this is asked per pair.
  private def implementsGeneralizedReducer(f: ReducibleFunction[_, _]): Boolean =
    f.getClass
      .getMethod("reducer", classOf[Array[V2Literal[_]]], classOf[ReducibleFunction[_, _]],
        classOf[Array[V2Literal[_]]])
      .getDeclaringClass != classOf[ReducibleFunction[_, _]]

Unimplemented would then be decided before the call rather than inferred from it, and an UnsupportedOperationException out of an implemented reducer would join Threw.

Counter-arguments, so this is a choice and not a request. Reflection on a connector class costs something, even memoized per class. A Scala object implementing the Java interface makes getDeclaringClass less obvious than it reads. And the UOE convention is what the other default methods in this package already use, so changing it only here makes the package inconsistent.

Fine to leave as is. If you do, the javadoc should say that an UnsupportedOperationException thrown from inside an implemented reducer is indistinguishable from not implementing it, since an implementor cannot otherwise guess that.

[metanil]

Yes, this was the known issue, rare but not impossible.
I didn't want to use reflection for it, but as you pointed out, there's already a history of using it, I might rethink this.

[peter-toth]

Finding 8. This one contradicts the test that uses it. The fixture doc says the dispatch "must not invoke (and log the throw from) the generalized overload once the deprecated one already produced a reducer", while the test "a throwing generalized overload is surfaced, not masked by the deprecated one" asserts the reverse: the generalized throw is authoritative and there is no fallback.

Three more sites still describe a shape that changed.

• KeyGroupedPartitioningSuite.scala:6343 says "the single-int dispatch tries reducer(int, ...) first (UOE), then the Literal[]". It has been generalized-first since the dispatch was redesigned, and BucketFunction overrides only the new overload, so no fallback is involved in that test at all.
• transformFunctions.scala:313, DualApiBucketFunction, says it verifies "that the dispatch falls back to the generalized overload when the deprecated one returns null". The deprecated overload is never consulted there, so its null is irrelevant, which is what the test body now says.
• transformFunctions.scala:576 and :595, StructBackedUDT and UdtParamFunction, describe noComplexLiteralParams as value-based and call the UDT "the case a DataType-based container check misses but a value-based one catches". The implemented guard is DataType-based and lists _: UserDefinedType[_] explicitly, so it catches it head-on.

22cd902c0d4 fixed a fourth site of this kind. These are the rest.

[peter-toth]

Finding 9. With this delegation the two defaults can call each other. A connector migrating to the generalized overload writes the natural shim for its zero-parameter transforms:

  @Override
  public Reducer<I, O> reducer(
      Literal<?>[] thisParams, ReducibleFunction<?, ?> other, Literal<?>[] otherParams) {
    if (thisParams.length == 0 && otherParams.length == 0) {
      return reducer(other);   // keep the old no-parameter logic
    }
    ...
  }

reducer(other) is not overridden, so it lands back here, which calls the override again. TransformExpression.reducers calls reducer(otherFunction) for exactly the zero-parameter pair, so the recursion is reachable, and the resulting StackOverflowError is a VirtualMachineError: Try's NonFatal does not catch it, so it takes the driver thread down rather than costing a shuffle.

An @implNote is enough, since the direction is not guessable from the signatures:

   * @implNote This default delegates to
   *     {@link #reducer(Literal[], ReducibleFunction, Literal[])}, so an implementation of that
   *     overload must not call back into this method.

[metanil]

Ya, I wanted to avoid adding implementation detail in javadoc, but good idea to add was part of @implNote. Thanks.

[metanil]

@peter-toth Just a note. Looks like Spark doesn't have @implNote build support. I'll add that one too in the build.
Let me know if we don't want to change build file in this PR.

[peter-toth]

Thanks for flagging it. That's on me: I suggested the tag without checking that the build doesn't register it. I'd rather not introduce a new Javadoc tag for a single note, so please drop the build changes and keep it as a plain paragraph instead:

   * <p>
   * This default delegates to
   * {@link #reducer(Literal[], ReducibleFunction, Literal[])}, so an implementation of that
   * overload must not call back into this method.

[metanil]

Thanks @peter-toth. I've addressed all of your findings except 4 (as per you instruction) and 7.
For 7, I'll follow up separately.

[peter-toth]

@metanil, #59165 is merged (master 05a20fb758b, branch-4.x d8d72d2d3b1), so you can rebase onto it now. A few notes, since this PR touches the same code:

• The guard sits in KeyedShuffleSpec.isExpressionCompatible now. A transform whose argument is not a bare column pairs only with a transform over the same argument shape (argumentShape). canCreatePartitioning refuses such a layout. createPartitioning and the identity reduce rebuild a transform through rebuiltOver, which asserts that the arguments are columns. TransformExpression.withReference is gone.
• With that gate in place, the total functionId from my withdrawn finding 4 is safe again: bucket(4, b + 1) can no longer pair with bucket(4, x), whatever the identity says. Keeping the Option is fine too, it's your call.
• This PR puts the literal parameters into children, so two of the new helpers need to skip them:
  • argumentsAreColumns checks children.forall(_.isInstanceOf[Attribute]). With a literal parameter, every bucket and truncate would count as a transform of an expression, so no side could be shuffled onto it any more.
  • rebuiltOver replaces every child with the other side's key, a literal parameter included. Your rewriteColumnSlots looks like the right replacement.
• Please keep argumentsAreColumns on bare Attributes rather than switching it to isColumnRef. A GetStructField argument must stay refused there: rebuilding bucket(4, s.a) over the other side's key gives bucket(4, s), which is the ClassCastException [SPARK-59887][SPARK-59901][SQL] Fix wrong results when a storage-partitioned join pairs a transform of a join-key expression with one of a column #59165 fixed.