peter-toth commented on code in PR #55885:
URL: https://github.com/apache/spark/pull/55885#discussion_r4122106752
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/AliasAwareOutputExpression.scala:
##########
@@ -89,6 +89,22 @@ trait PartitioningPreservingUnaryExecNode extends
UnaryExecNode
}
}
+ /**
+ * Projects a partitioning expression through the output aliases. For a
[[TransformExpression]]
+ * only the column argument is projected: `aliasMap` also maps aliased
literals, so `2 AS w` would
+ * otherwise turn `truncate(data, 2)` into `truncate(d, w)`.
+ */
+ private def projectPartitionExpression(expr: Expression):
LazyList[Expression] = expr match {
Review Comment:
**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:
```scala
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`:
```scala
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:
```scala
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala:
##########
@@ -6206,6 +6211,655 @@ class KeyGroupedPartitioningSuite
}
}
+ test("SPARK-50593: cross-function truncate vs bucket should NOT trigger
SPJ") {
+ val partitions1 = Array(
+ Expressions.apply("truncate", Expressions.column("data"),
Expressions.literal(3))
+ )
+ val partitions2 = Array(
+ Expressions.bucket(4, "data")
+ )
+
+ createTable("trunc_cross1", columns, partitions1)
+ sql("INSERT INTO testcat.ns.trunc_cross1 VALUES " +
+ "(0, 'aaa', CAST('2022-01-01' AS timestamp)), " +
+ "(1, 'bbb', CAST('2021-01-01' AS timestamp))")
+
+ createTable("trunc_cross2", columns2, partitions2)
+ sql("INSERT INTO testcat.ns.trunc_cross2 VALUES " +
+ "(1, 5, 'aaa'), " +
+ "(5, 10, 'bbb')")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+
+ val df = sql(
+ s"""
+ |${selectWithMergeJoinHint("trunc_cross1", "trunc_cross2")}
+ |trunc_cross1.id, trunc_cross2.store_id
+ |FROM testcat.ns.trunc_cross1 JOIN testcat.ns.trunc_cross2
+ |ON trunc_cross1.data = trunc_cross2.data
+ |ORDER BY trunc_cross1.id
+ |""".stripMargin)
+
+ // Different functions (truncate vs bucket) are not mutually reducible,
so a shuffle
+ // must still be planned.
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.nonEmpty,
+ "truncate vs bucket are not compatible - a shuffle should be present,
" +
+ "but none was planned")
+ checkAnswer(df, Seq(Row(0, 1), Row(1, 5)))
+ }
+ }
+
+ test("SPARK-50593: truncate(3) vs truncate(5) triggers SPJ via width
reducer") {
+ // Exercises the Literal[]-based reducer path end-to-end: truncate widths
3 and 5
+ // are mutually reducible (reduce the larger to the smaller), so SPJ must
avoid the shuffle.
+ val table1 = "trunc_three"
+ val table2 = "trunc_five"
+
+ val partitions1 = Array(
+ Expressions.apply("truncate", Expressions.column("data"),
Expressions.literal(3)))
+ val partitions2 = Array(
+ Expressions.apply("truncate", Expressions.column("data"),
Expressions.literal(5)))
+
+ createTable(table1, columns, partitions1)
+ 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, partitions2)
+ 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))")
+
+ withSQLConf(
+ SQLConf.V2_BUCKETING_PUSH_PART_VALUES_ENABLED.key -> "true",
+ SQLConf.V2_BUCKETING_ALLOW_COMPATIBLE_TRANSFORMS.key -> "true") {
+
+ 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)
+
+ val shuffles = collectShuffles(df.queryExecution.executedPlan)
+ assert(shuffles.isEmpty,
+ "truncate(3) vs truncate(5) should avoid shuffle via the width
reducer, " +
+ "but a shuffle was planned")
+ checkAnswer(df, Seq(Row(0, 10), Row(1, 20), Row(2, 30)))
Review Comment:
**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`.
```scala
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.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -133,18 +182,178 @@ case class TransformExpression(
def withReference(attr: Attribute): TransformExpression =
transform { case _: AttributeReference => attr
}.asInstanceOf[TransformExpression]
- // Return a Reducer for a reducible function on another reducible function
- private def reducer(
- thisFunction: ReducibleFunction[_, _],
- thisNumBucketsOpt: Option[Int],
- otherFunction: ReducibleFunction[_, _],
- otherNumBucketsOpt: Option[Int]): Option[Reducer[_, _]] = {
- val res = (thisNumBucketsOpt, otherNumBucketsOpt) match {
- case (Some(numBuckets), Some(otherNumBuckets)) =>
- thisFunction.reducer(numBuckets, otherFunction, otherNumBuckets)
- case _ => thisFunction.reducer(otherFunction)
+ /**
+ * Rewrites this transform's column arguments with `rewriteColumn`, leaving
its literal parameters
+ * untouched. Unlike [[withReference]], which swaps the attribute *inside* a
column argument (so a
+ * `GetStructField` path above it is kept), this replaces the whole argument.
+ *
+ * Since literal parameters (the bucket count, the truncate width) live in
`children` rather than
+ * in a field of their own, any path that rewrites a transform's children
has to skip them. A
+ * parameter is not an aliasable value: substituting it changes what the
transform computes, drops
+ * it out of `functionId`, adds a second entry to `references` (tripping
+ * `KeyedShuffleSpec.keyPositions`' single-reference assert), and makes
+ * `KeyedPartitioning.supportsExpressions` reject the partitioning outright
-- so SPJ is silently
+ * lost. Every such path should go through here rather than reimplement the
skip.
+ *
+ * Callers: `KeyedShuffleSpec.createPartitioning`, which retargets a
transform at the other side's
+ * clustering key, and `PartitioningPreservingUnaryExecNode`, which
retargets it at an aliased
+ * output attribute.
+ */
+ def rewriteColumnSlots(rewriteColumn: Expression => Expression):
TransformExpression =
+ copy(children = children.map {
+ case l: Literal => l
+ case c => rewriteColumn(c)
+ })
+
+ /** The non-literal arguments. */
+ def columnSlots: Seq[Expression] =
children.filterNot(_.isInstanceOf[Literal])
+
+ /**
+ * Extract all literal parameters of this transform as V2 [[V2Literal]]s,
preserving each value's
+ * internal representation and its `DataType`. Only consulted once a reducer
path has confirmed
+ * the literal params already match the declared input types (see
+ * [[literalParamsMatchInputTypes]]), so no type coercion happens here.
Memoized.
+ *
+ * Examples:
+ * bucket(4, col) => [Literal(4, IntegerType)]
+ * truncate(col, 3) => [Literal(3, IntegerType)]
+ * days(col) => [] (no literals)
+ */
+ private lazy val extractParameters: Array[V2Literal[_]] =
+ literalChildren.map(l => LiteralValue(l.value, l.dataType):
V2Literal[_]).toArray
+
+ /**
+ * Whether the `select`ed children match the bound function's declared input
type at their
+ * positions. A child beyond the declared arity has no declared type to
compare against (e.g. an
+ * arity-flexible function), so it is left to the connector reducer / other
guards. The DataType
+ * match is exact by design: any mismatch (including cosmetic ones like
Array `containsNull` or
+ * Decimal precision/scale) fails safe to a shuffle. See the two predicates
below for the callers.
+ *
+ * Reading `dataType` can throw rather than answer. [[withReference]]
retargets a transform at
+ * another key, and a [[GetStructField]] column slot retargeted at a
non-struct attribute -- an
+ * `identity` column on the other side of a join, say -- leaves
`GetStructField(intCol, 0)`, whose
+ * `dataType` raises `INTERNAL_ERROR: GetStructField requires a StructType
child`. Both callers
+ * are gates that must answer "not reducible" rather than fail the query, so
a child whose type
+ * cannot be read is treated as not matching.
+ */
+ private def inputTypesMatch(select: Expression => Boolean): Boolean = {
+ val declaredTypes = function.inputTypes()
+ children.zipWithIndex.forall {
+ case (c, i) =>
+ !select(c) || i >= declaredTypes.length ||
+ Try(c.dataType).toOption.contains(declaredTypes(i))
+ }
+ }
+
+ /**
+ * Whether every literal parameter matches its declared input type. Used by
the transform-vs-
+ * transform reducer path, which hands literal *values* to the connector
reducer without Analyzer
+ * type coercion. A literal whose type differs from the declared input type
(a legal implicit cast
+ * under [[BoundFunction]]) is not reducible: the join falls back to a
shuffle rather than handing
+ * the connector a value the partitions were not built on, which it would
then mis-cast. Column
+ * slots are not checked (they are not passed to the connector); the eval
path uses
+ * [[argsMatchInputTypes]] instead.
+ */
+ lazy val literalParamsMatchInputTypes: Boolean =
inputTypesMatch(_.isInstanceOf[Literal])
+
+ /**
+ * Whether every argument has its declared type, at exactly the declared
arity. Required before
+ * evaluating the transform directly: no cast is applied, and a mismatch
would fail at evaluation.
+ */
+ lazy val argsMatchInputTypes: Boolean =
+ children.length == function.inputTypes().length && inputTypesMatch(_ =>
true)
+
+ /**
+ * Whether each position holds a literal on both sides or a column reference
on both sides.
+ * Literal values and arity may differ; the connector reducer decides those.
+ */
+ private def sameArgumentLayout(other: TransformExpression): Boolean =
+ children.zip(other.children).forall {
+ case (_: Literal, _: Literal) => true
+ case (c1, c2) => TransformExpression.isColumnRef(c1) &&
TransformExpression.isColumnRef(c2)
+ }
+
+ /**
+ * Whether no literal parameter has an array, map, struct or UDT type; those
are not passed to a
+ * connector reducer. Other scalars, such as intervals, are.
+ */
+ private def noComplexLiteralParams: Boolean =
+ literalChildren.forall(_.dataType match {
+ case _: ArrayType | _: MapType | _: StructType | _: UserDefinedType[_]
=> false
+ case _ => true
+ })
+
+ /**
+ * Return a [[Reducer]] that maps this transform's partition keys onto
`other`'s, or None if there
+ * is none: when either function is not a [[ReducibleFunction]], when the
argument layouts or
+ * literal parameters rule a reducer out, or when the connector reports it
is not reducible.
+ * Handles both parameterized (bucket, truncate) and non-parameterized
(days, hours) functions.
+ */
+ def reducers(other: TransformExpression): Option[Reducer[_, _]] = {
+ import TransformExpression._
+ val thisFunction: ReducibleFunction[_, _] = function match {
+ case f: ReducibleFunction[_, _] => f
+ case _ => return None
+ }
+ val otherFunction: ReducibleFunction[_, _] = other.function match {
+ case o: ReducibleFunction[_, _] => o
+ case _ => return None
+ }
+ if (!sameArgumentLayout(other) ||
+ !literalParamsMatchInputTypes || !other.literalParamsMatchInputTypes ||
+ !noComplexLiteralParams || !other.noComplexLiteralParams) {
+ return None
+ }
+
+ val thisParams = extractParameters
+ val otherParams = other.extractParameters
+ val thisName = function.canonicalName()
+
+ // Only a single non-null IntegerType parameter per side may use the
deprecated int overload; a
+ // typed null would otherwise be read as 0.
+ def isSingleInt(p: Array[V2Literal[_]]): Boolean = {
+ p.length == 1 && p(0).dataType == IntegerType && p(0).value() != null
+ }
+
+ // Probe one reducer overload into an Outcome. Pure -- logging is decided
once, below.
+ def probe(call: => Reducer[_, _]): Outcome = Try(Option(call)) match {
+ case Success(Some(r)) => Reducible(r)
+ case Success(None) => NotReducible
+ case Failure(_: UnsupportedOperationException) => Unimplemented
+ case Failure(e) => Threw(e)
+ }
+
+ // Prefer the generalized overload. Use the deprecated one only if the
generalized one is not
+ // implemented; any other outcome is final.
+ val outcome =
+ if (thisParams.isEmpty && otherParams.isEmpty) {
+ probe(thisFunction.reducer(otherFunction))
+ } else {
+ probe(thisFunction.reducer(thisParams, otherFunction, otherParams))
match {
+ // The deprecated overload is documented for bucket against bucket,
so only offer it a
+ // pair of the same function.
+ case Unimplemented if isSingleInt(thisParams) &&
isSingleInt(otherParams) &&
+ function.name() == other.function.name() =>
Review Comment:
**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.
```suggestion
function.canonicalName() == other.function.canonicalName() =>
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -133,18 +182,178 @@ case class TransformExpression(
def withReference(attr: Attribute): TransformExpression =
transform { case _: AttributeReference => attr
}.asInstanceOf[TransformExpression]
- // Return a Reducer for a reducible function on another reducible function
- private def reducer(
- thisFunction: ReducibleFunction[_, _],
- thisNumBucketsOpt: Option[Int],
- otherFunction: ReducibleFunction[_, _],
- otherNumBucketsOpt: Option[Int]): Option[Reducer[_, _]] = {
- val res = (thisNumBucketsOpt, otherNumBucketsOpt) match {
- case (Some(numBuckets), Some(otherNumBuckets)) =>
- thisFunction.reducer(numBuckets, otherFunction, otherNumBuckets)
- case _ => thisFunction.reducer(otherFunction)
+ /**
+ * Rewrites this transform's column arguments with `rewriteColumn`, leaving
its literal parameters
+ * untouched. Unlike [[withReference]], which swaps the attribute *inside* a
column argument (so a
+ * `GetStructField` path above it is kept), this replaces the whole argument.
+ *
+ * Since literal parameters (the bucket count, the truncate width) live in
`children` rather than
+ * in a field of their own, any path that rewrites a transform's children
has to skip them. A
+ * parameter is not an aliasable value: substituting it changes what the
transform computes, drops
+ * it out of `functionId`, adds a second entry to `references` (tripping
+ * `KeyedShuffleSpec.keyPositions`' single-reference assert), and makes
+ * `KeyedPartitioning.supportsExpressions` reject the partitioning outright
-- so SPJ is silently
+ * lost. Every such path should go through here rather than reimplement the
skip.
+ *
+ * Callers: `KeyedShuffleSpec.createPartitioning`, which retargets a
transform at the other side's
+ * clustering key, and `PartitioningPreservingUnaryExecNode`, which
retargets it at an aliased
+ * output attribute.
+ */
+ def rewriteColumnSlots(rewriteColumn: Expression => Expression):
TransformExpression =
+ copy(children = children.map {
+ case l: Literal => l
+ case c => rewriteColumn(c)
+ })
+
+ /** The non-literal arguments. */
+ def columnSlots: Seq[Expression] =
children.filterNot(_.isInstanceOf[Literal])
+
+ /**
+ * Extract all literal parameters of this transform as V2 [[V2Literal]]s,
preserving each value's
+ * internal representation and its `DataType`. Only consulted once a reducer
path has confirmed
+ * the literal params already match the declared input types (see
+ * [[literalParamsMatchInputTypes]]), so no type coercion happens here.
Memoized.
+ *
+ * Examples:
+ * bucket(4, col) => [Literal(4, IntegerType)]
+ * truncate(col, 3) => [Literal(3, IntegerType)]
+ * days(col) => [] (no literals)
+ */
+ private lazy val extractParameters: Array[V2Literal[_]] =
+ literalChildren.map(l => LiteralValue(l.value, l.dataType):
V2Literal[_]).toArray
+
+ /**
+ * Whether the `select`ed children match the bound function's declared input
type at their
+ * positions. A child beyond the declared arity has no declared type to
compare against (e.g. an
+ * arity-flexible function), so it is left to the connector reducer / other
guards. The DataType
+ * match is exact by design: any mismatch (including cosmetic ones like
Array `containsNull` or
+ * Decimal precision/scale) fails safe to a shuffle. See the two predicates
below for the callers.
+ *
+ * Reading `dataType` can throw rather than answer. [[withReference]]
retargets a transform at
+ * another key, and a [[GetStructField]] column slot retargeted at a
non-struct attribute -- an
+ * `identity` column on the other side of a join, say -- leaves
`GetStructField(intCol, 0)`, whose
+ * `dataType` raises `INTERNAL_ERROR: GetStructField requires a StructType
child`. Both callers
+ * are gates that must answer "not reducible" rather than fail the query, so
a child whose type
+ * cannot be read is treated as not matching.
+ */
+ private def inputTypesMatch(select: Expression => Boolean): Boolean = {
+ val declaredTypes = function.inputTypes()
+ children.zipWithIndex.forall {
+ case (c, i) =>
+ !select(c) || i >= declaredTypes.length ||
+ Try(c.dataType).toOption.contains(declaredTypes(i))
+ }
+ }
+
+ /**
+ * Whether every literal parameter matches its declared input type. Used by
the transform-vs-
+ * transform reducer path, which hands literal *values* to the connector
reducer without Analyzer
+ * type coercion. A literal whose type differs from the declared input type
(a legal implicit cast
+ * under [[BoundFunction]]) is not reducible: the join falls back to a
shuffle rather than handing
+ * the connector a value the partitions were not built on, which it would
then mis-cast. Column
+ * slots are not checked (they are not passed to the connector); the eval
path uses
+ * [[argsMatchInputTypes]] instead.
+ */
+ lazy val literalParamsMatchInputTypes: Boolean =
inputTypesMatch(_.isInstanceOf[Literal])
+
+ /**
+ * Whether every argument has its declared type, at exactly the declared
arity. Required before
+ * evaluating the transform directly: no cast is applied, and a mismatch
would fail at evaluation.
+ */
+ lazy val argsMatchInputTypes: Boolean =
+ children.length == function.inputTypes().length && inputTypesMatch(_ =>
true)
+
+ /**
+ * Whether each position holds a literal on both sides or a column reference
on both sides.
+ * Literal values and arity may differ; the connector reducer decides those.
+ */
+ private def sameArgumentLayout(other: TransformExpression): Boolean =
+ children.zip(other.children).forall {
+ case (_: Literal, _: Literal) => true
+ case (c1, c2) => TransformExpression.isColumnRef(c1) &&
TransformExpression.isColumnRef(c2)
+ }
+
+ /**
+ * Whether no literal parameter has an array, map, struct or UDT type; those
are not passed to a
+ * connector reducer. Other scalars, such as intervals, are.
+ */
+ private def noComplexLiteralParams: Boolean =
+ literalChildren.forall(_.dataType match {
+ case _: ArrayType | _: MapType | _: StructType | _: UserDefinedType[_]
=> false
+ case _ => true
+ })
+
+ /**
+ * Return a [[Reducer]] that maps this transform's partition keys onto
`other`'s, or None if there
+ * is none: when either function is not a [[ReducibleFunction]], when the
argument layouts or
+ * literal parameters rule a reducer out, or when the connector reports it
is not reducible.
+ * Handles both parameterized (bucket, truncate) and non-parameterized
(days, hours) functions.
+ */
+ def reducers(other: TransformExpression): Option[Reducer[_, _]] = {
+ import TransformExpression._
+ val thisFunction: ReducibleFunction[_, _] = function match {
+ case f: ReducibleFunction[_, _] => f
+ case _ => return None
+ }
+ val otherFunction: ReducibleFunction[_, _] = other.function match {
+ case o: ReducibleFunction[_, _] => o
+ case _ => return None
+ }
+ if (!sameArgumentLayout(other) ||
+ !literalParamsMatchInputTypes || !other.literalParamsMatchInputTypes ||
+ !noComplexLiteralParams || !other.noComplexLiteralParams) {
+ return None
+ }
+
+ val thisParams = extractParameters
+ val otherParams = other.extractParameters
+ val thisName = function.canonicalName()
+
+ // Only a single non-null IntegerType parameter per side may use the
deprecated int overload; a
+ // typed null would otherwise be read as 0.
+ def isSingleInt(p: Array[V2Literal[_]]): Boolean = {
+ p.length == 1 && p(0).dataType == IntegerType && p(0).value() != null
+ }
+
+ // Probe one reducer overload into an Outcome. Pure -- logging is decided
once, below.
+ def probe(call: => Reducer[_, _]): Outcome = Try(Option(call)) match {
+ case Success(Some(r)) => Reducible(r)
+ case Success(None) => NotReducible
+ case Failure(_: UnsupportedOperationException) => Unimplemented
Review Comment:
**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:
```scala
// 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.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/functions/ReducibleFunction.java:
##########
@@ -101,6 +155,6 @@ default Reducer<I, O> reducer(
* @return a reduction function if it is reducible, null if not.
*/
default Reducer<I, O> reducer(ReducibleFunction<?, ?> otherFunction) {
- throw new UnsupportedOperationException();
+ return reducer(new Literal<?>[0], otherFunction, new Literal<?>[0]);
Review Comment:
**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:
```java
@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.
```
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -52,23 +87,46 @@ case class TransformFunctionId(canonicalName: String,
numBucketsOpt: Option[Int]
case class TransformExpression(
function: BoundFunction,
children: Seq[Expression],
- numBucketsOpt: Option[Int] = None,
- reducedWith: Option[TransformFunctionId] = None) extends Expression {
+ reducedWith: Option[TransformFunctionId] = None) extends Expression with
Logging {
override def nullable: Boolean = true
/** Drops the trailing `reducedWith` when it is unset, so the ordinary form
is unchanged. */
override protected def stringArgs: Iterator[Any] =
if (reducedWith.isEmpty) super.stringArgs.take(productArity - 1) else
super.stringArgs
- /** The identity of this expression's transform function. */
- lazy val functionId: TransformFunctionId =
- TransformFunctionId(function.canonicalName(), numBucketsOpt)
+ /**
+ * Extract literal children (constant parameters) from this transform. These
are constant
+ * arguments like the bucket count in `bucket(n, col)` or the width in
`truncate(col, width)`.
+ * Positions are dropped, so this is for handing parameters to a connector
reducer, NOT for
+ * identity -- see [[functionId]] and [[TransformFunctionId.argumentShapes]].
+ */
+ private lazy val literalChildren: Seq[Literal] =
+ children.collect { case l: Literal => l }
/**
- * Whether this [[TransformExpression]] has the same semantics as `other`.
- * For instance, `bucket(32, c)` is equal to `bucket(32, d)`, but not to
`bucket(16, d)` or
- * `year(c)`.
+ * This transform's identity, or None if some argument is not a literal, a
nested transform or a
+ * column reference (e.g. `c + 1`); such a transform is not the same as any
other.
+ */
+ lazy val functionId: Option[TransformFunctionId] = {
+ import TransformFunctionId.ArgumentShape
+ val shapes = children.map {
+ case l: Literal => Some(ArgumentShape.Param(l))
+ case t: TransformExpression => t.functionId.map(ArgumentShape.Nested(_))
+ case c if TransformExpression.isColumnRef(c) =>
Some(ArgumentShape.Column)
Review Comment:
**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:
```scala
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
I have this and finding 1 as one patch against `18f6f8f` (75 insertions, 78
deletions over 6 files, including the test adjustments). Happy to push it to a
branch on my fork for you to cherry-pick if that is easier than redoing it --
just say so.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/TransformExpression.scala:
##########
@@ -133,18 +182,178 @@ case class TransformExpression(
def withReference(attr: Attribute): TransformExpression =
transform { case _: AttributeReference => attr
}.asInstanceOf[TransformExpression]
- // Return a Reducer for a reducible function on another reducible function
- private def reducer(
- thisFunction: ReducibleFunction[_, _],
- thisNumBucketsOpt: Option[Int],
- otherFunction: ReducibleFunction[_, _],
- otherNumBucketsOpt: Option[Int]): Option[Reducer[_, _]] = {
- val res = (thisNumBucketsOpt, otherNumBucketsOpt) match {
- case (Some(numBuckets), Some(otherNumBuckets)) =>
- thisFunction.reducer(numBuckets, otherFunction, otherNumBuckets)
- case _ => thisFunction.reducer(otherFunction)
+ /**
+ * Rewrites this transform's column arguments with `rewriteColumn`, leaving
its literal parameters
+ * untouched. Unlike [[withReference]], which swaps the attribute *inside* a
column argument (so a
+ * `GetStructField` path above it is kept), this replaces the whole argument.
+ *
+ * Since literal parameters (the bucket count, the truncate width) live in
`children` rather than
+ * in a field of their own, any path that rewrites a transform's children
has to skip them. A
+ * parameter is not an aliasable value: substituting it changes what the
transform computes, drops
+ * it out of `functionId`, adds a second entry to `references` (tripping
+ * `KeyedShuffleSpec.keyPositions`' single-reference assert), and makes
+ * `KeyedPartitioning.supportsExpressions` reject the partitioning outright
-- so SPJ is silently
+ * lost. Every such path should go through here rather than reimplement the
skip.
+ *
+ * Callers: `KeyedShuffleSpec.createPartitioning`, which retargets a
transform at the other side's
+ * clustering key, and `PartitioningPreservingUnaryExecNode`, which
retargets it at an aliased
+ * output attribute.
+ */
+ def rewriteColumnSlots(rewriteColumn: Expression => Expression):
TransformExpression =
+ copy(children = children.map {
+ case l: Literal => l
+ case c => rewriteColumn(c)
+ })
+
+ /** The non-literal arguments. */
+ def columnSlots: Seq[Expression] =
children.filterNot(_.isInstanceOf[Literal])
+
+ /**
+ * Extract all literal parameters of this transform as V2 [[V2Literal]]s,
preserving each value's
+ * internal representation and its `DataType`. Only consulted once a reducer
path has confirmed
+ * the literal params already match the declared input types (see
+ * [[literalParamsMatchInputTypes]]), so no type coercion happens here.
Memoized.
+ *
+ * Examples:
+ * bucket(4, col) => [Literal(4, IntegerType)]
+ * truncate(col, 3) => [Literal(3, IntegerType)]
+ * days(col) => [] (no literals)
+ */
+ private lazy val extractParameters: Array[V2Literal[_]] =
+ literalChildren.map(l => LiteralValue(l.value, l.dataType):
V2Literal[_]).toArray
+
+ /**
+ * Whether the `select`ed children match the bound function's declared input
type at their
+ * positions. A child beyond the declared arity has no declared type to
compare against (e.g. an
+ * arity-flexible function), so it is left to the connector reducer / other
guards. The DataType
+ * match is exact by design: any mismatch (including cosmetic ones like
Array `containsNull` or
+ * Decimal precision/scale) fails safe to a shuffle. See the two predicates
below for the callers.
+ *
+ * Reading `dataType` can throw rather than answer. [[withReference]]
retargets a transform at
+ * another key, and a [[GetStructField]] column slot retargeted at a
non-struct attribute -- an
+ * `identity` column on the other side of a join, say -- leaves
`GetStructField(intCol, 0)`, whose
+ * `dataType` raises `INTERNAL_ERROR: GetStructField requires a StructType
child`. Both callers
+ * are gates that must answer "not reducible" rather than fail the query, so
a child whose type
+ * cannot be read is treated as not matching.
+ */
+ private def inputTypesMatch(select: Expression => Boolean): Boolean = {
+ val declaredTypes = function.inputTypes()
+ children.zipWithIndex.forall {
+ case (c, i) =>
+ !select(c) || i >= declaredTypes.length ||
+ Try(c.dataType).toOption.contains(declaredTypes(i))
+ }
+ }
+
+ /**
+ * Whether every literal parameter matches its declared input type. Used by
the transform-vs-
+ * transform reducer path, which hands literal *values* to the connector
reducer without Analyzer
+ * type coercion. A literal whose type differs from the declared input type
(a legal implicit cast
+ * under [[BoundFunction]]) is not reducible: the join falls back to a
shuffle rather than handing
+ * the connector a value the partitions were not built on, which it would
then mis-cast. Column
+ * slots are not checked (they are not passed to the connector); the eval
path uses
+ * [[argsMatchInputTypes]] instead.
+ */
+ lazy val literalParamsMatchInputTypes: Boolean =
inputTypesMatch(_.isInstanceOf[Literal])
+
+ /**
+ * Whether every argument has its declared type, at exactly the declared
arity. Required before
+ * evaluating the transform directly: no cast is applied, and a mismatch
would fail at evaluation.
+ */
+ lazy val argsMatchInputTypes: Boolean =
+ children.length == function.inputTypes().length && inputTypesMatch(_ =>
true)
+
+ /**
+ * Whether each position holds a literal on both sides or a column reference
on both sides.
+ * Literal values and arity may differ; the connector reducer decides those.
+ */
+ private def sameArgumentLayout(other: TransformExpression): Boolean =
+ children.zip(other.children).forall {
+ case (_: Literal, _: Literal) => true
+ case (c1, c2) => TransformExpression.isColumnRef(c1) &&
TransformExpression.isColumnRef(c2)
+ }
+
+ /**
+ * Whether no literal parameter has an array, map, struct or UDT type; those
are not passed to a
+ * connector reducer. Other scalars, such as intervals, are.
+ */
+ private def noComplexLiteralParams: Boolean =
+ literalChildren.forall(_.dataType match {
+ case _: ArrayType | _: MapType | _: StructType | _: UserDefinedType[_]
=> false
+ case _ => true
+ })
+
+ /**
+ * Return a [[Reducer]] that maps this transform's partition keys onto
`other`'s, or None if there
+ * is none: when either function is not a [[ReducibleFunction]], when the
argument layouts or
+ * literal parameters rule a reducer out, or when the connector reports it
is not reducible.
+ * Handles both parameterized (bucket, truncate) and non-parameterized
(days, hours) functions.
+ */
+ def reducers(other: TransformExpression): Option[Reducer[_, _]] = {
+ import TransformExpression._
+ val thisFunction: ReducibleFunction[_, _] = function match {
+ case f: ReducibleFunction[_, _] => f
+ case _ => return None
+ }
+ val otherFunction: ReducibleFunction[_, _] = other.function match {
+ case o: ReducibleFunction[_, _] => o
+ case _ => return None
+ }
+ if (!sameArgumentLayout(other) ||
+ !literalParamsMatchInputTypes || !other.literalParamsMatchInputTypes ||
+ !noComplexLiteralParams || !other.noComplexLiteralParams) {
+ return None
+ }
+
+ val thisParams = extractParameters
+ val otherParams = other.extractParameters
+ val thisName = function.canonicalName()
+
+ // Only a single non-null IntegerType parameter per side may use the
deprecated int overload; a
+ // typed null would otherwise be read as 0.
+ def isSingleInt(p: Array[V2Literal[_]]): Boolean = {
+ p.length == 1 && p(0).dataType == IntegerType && p(0).value() != null
Review Comment:
**Finding 3.** This clause keeps a typed null out of the deprecated
overload, which is what
[r3503138262](https://github.com/apache/spark/pull/55885#discussion_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:
```scala
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/functions/transformFunctions.scala:
##########
@@ -279,20 +409,372 @@ object SignedZerosFunction extends
ScalarFunction[Double] {
}
object UnboundTruncateFunction extends UnboundFunction {
- override def bind(inputType: StructType): BoundFunction = TruncateFunction
+ override def bind(inputType: StructType): BoundFunction = {
+ if (inputType.size == 2) {
+ inputType.head.dataType match {
+ case StringType => TruncateFunction
+ case IntegerType => IntegerTruncateFunction
+ case _ =>
+ throw new UnsupportedOperationException(
+ s"'truncate' does not support data type:
${inputType.head.dataType}")
+ }
+ } else {
+ throw new UnsupportedOperationException(
+ "'truncate' requires exactly 2 arguments: (column, width)")
+ }
+ }
+
override def description(): String = name()
override def name(): String = "truncate"
}
-object TruncateFunction extends ScalarFunction[UTF8String] {
+/**
+ * Truncate transform for String type.
+ * Follows Iceberg spec: truncate(str, L) = str[0:L]
+ *
+ * Implements ReducibleFunction: ANY two different widths are compatible.
+ * The reducer uses the smaller width.
+ */
+object TruncateFunction
+ extends ScalarFunction[UTF8String]
+ with ReducibleFunction[UTF8String, UTF8String] {
override def inputTypes(): Array[DataType] = Array(StringType, IntegerType)
override def resultType(): DataType = StringType
override def name(): String = "truncate"
override def canonicalName(): String = name()
override def toString: String = name()
override def produceResult(input: InternalRow): UTF8String = {
val str = input.getUTF8String(0)
- val length = input.getInt(1)
- str.substring(0, length)
+ val width = input.getInt(1)
+ str.substring(0, width)
+ }
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[UTF8String, UTF8String] = {
+
+ if (otherFunc == TruncateFunction && thisParams.length == 1 &&
otherParams.length == 1) {
+ val thisWidth = thisParams(0).value().asInstanceOf[Int]
+ val otherWidth = otherParams(0).value().asInstanceOf[Int]
+ val smallerWidth = math.min(thisWidth, otherWidth)
+
+ if (smallerWidth != thisWidth) {
+ return TruncateReducer(smallerWidth)
+ }
+ }
+ null
+ }
+}
+
+case class TruncateReducer(width: Int) extends Reducer[UTF8String, UTF8String]
{
+ override def reduce(value: UTF8String): UTF8String = {
+ value.substring(0, width)
+ }
+ override def resultType(): DataType = StringType
+ override def displayName(): String = s"truncate($width)"
+}
+
+/**
+ * Truncate transform for Integer type.
+ * Follows Iceberg spec: truncate(value, W) = value - (((value % W) + W) % W),
which snaps `value`
+ * down to a multiple of `W`.
+ *
+ * Implements ReducibleFunction: truncate(v, W1) and truncate(v, W2) are
always reducible onto a
+ * common coarser grid of multiples of lcm(W1, W2). The finer side (whose
width does not already
+ * equal the lcm) reduces by snapping to that grid; when W2 is a multiple of
W1 the lcm is simply
+ * the coarser width W2.
+ */
+object IntegerTruncateFunction
+ extends ScalarFunction[Int]
+ with ReducibleFunction[Int, Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType, IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "truncate"
+ override def canonicalName(): String = name()
Review Comment:
**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.
```suggestion
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/functions/transformFunctions.scala:
##########
@@ -279,20 +409,372 @@ object SignedZerosFunction extends
ScalarFunction[Double] {
}
object UnboundTruncateFunction extends UnboundFunction {
- override def bind(inputType: StructType): BoundFunction = TruncateFunction
+ override def bind(inputType: StructType): BoundFunction = {
+ if (inputType.size == 2) {
+ inputType.head.dataType match {
+ case StringType => TruncateFunction
+ case IntegerType => IntegerTruncateFunction
+ case _ =>
+ throw new UnsupportedOperationException(
+ s"'truncate' does not support data type:
${inputType.head.dataType}")
+ }
+ } else {
+ throw new UnsupportedOperationException(
+ "'truncate' requires exactly 2 arguments: (column, width)")
+ }
+ }
+
override def description(): String = name()
override def name(): String = "truncate"
}
-object TruncateFunction extends ScalarFunction[UTF8String] {
+/**
+ * Truncate transform for String type.
+ * Follows Iceberg spec: truncate(str, L) = str[0:L]
+ *
+ * Implements ReducibleFunction: ANY two different widths are compatible.
+ * The reducer uses the smaller width.
+ */
+object TruncateFunction
+ extends ScalarFunction[UTF8String]
+ with ReducibleFunction[UTF8String, UTF8String] {
override def inputTypes(): Array[DataType] = Array(StringType, IntegerType)
override def resultType(): DataType = StringType
override def name(): String = "truncate"
override def canonicalName(): String = name()
override def toString: String = name()
override def produceResult(input: InternalRow): UTF8String = {
val str = input.getUTF8String(0)
- val length = input.getInt(1)
- str.substring(0, length)
+ val width = input.getInt(1)
+ str.substring(0, width)
+ }
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[UTF8String, UTF8String] = {
+
+ if (otherFunc == TruncateFunction && thisParams.length == 1 &&
otherParams.length == 1) {
+ val thisWidth = thisParams(0).value().asInstanceOf[Int]
+ val otherWidth = otherParams(0).value().asInstanceOf[Int]
+ val smallerWidth = math.min(thisWidth, otherWidth)
+
+ if (smallerWidth != thisWidth) {
+ return TruncateReducer(smallerWidth)
+ }
+ }
+ null
+ }
+}
+
+case class TruncateReducer(width: Int) extends Reducer[UTF8String, UTF8String]
{
+ override def reduce(value: UTF8String): UTF8String = {
+ value.substring(0, width)
+ }
+ override def resultType(): DataType = StringType
+ override def displayName(): String = s"truncate($width)"
+}
+
+/**
+ * Truncate transform for Integer type.
+ * Follows Iceberg spec: truncate(value, W) = value - (((value % W) + W) % W),
which snaps `value`
+ * down to a multiple of `W`.
+ *
+ * Implements ReducibleFunction: truncate(v, W1) and truncate(v, W2) are
always reducible onto a
+ * common coarser grid of multiples of lcm(W1, W2). The finer side (whose
width does not already
+ * equal the lcm) reduces by snapping to that grid; when W2 is a multiple of
W1 the lcm is simply
+ * the coarser width W2.
+ */
+object IntegerTruncateFunction
+ extends ScalarFunction[Int]
+ with ReducibleFunction[Int, Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType, IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "truncate"
+ override def canonicalName(): String = name()
+ override def toString: String = name()
+ override def produceResult(input: InternalRow): Int = {
+ val value = input.getInt(0)
+ val width = input.getInt(1)
+ value - (((value % width) + width) % width)
+ }
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[Int, Int] = {
+ if (otherFunc == IntegerTruncateFunction && thisParams.length == 1 &&
otherParams.length == 1) {
+ val thisWidth = thisParams(0).value().asInstanceOf[Int]
+ val otherWidth = otherParams(0).value().asInstanceOf[Int]
+ val common = lcm(thisWidth, otherWidth)
+ // Only the finer side reduces; if `common == thisWidth` this side is
already the common grid.
+ if (common != thisWidth) {
+ return IntTruncateReducer(common)
+ }
+ }
+ null
+ }
+
+ private def lcm(a: Int, b: Int): Int = {
+ val g = BigInt(a).gcd(BigInt(b))
+ (BigInt(a) / g * BigInt(b)).toInt
+ }
+}
+
+case class IntTruncateReducer(width: Int) extends Reducer[Int, Int] {
+ override def reduce(value: Int): Int = value - (((value % width) + width) %
width)
+ override def resultType(): DataType = IntegerType
+ override def displayName(): String = s"truncate($width)"
+}
+
+/**
+ * A transform whose reducer is defined across a zero-parameter vs
one-parameter shape, e.g.
+ * `zero_or_one(col)` reducing onto `zero_or_one(col, 2)`. Used to verify the
dispatch does not
+ * globally require equal child counts before invoking the reducer.
+ */
+object ZeroOrOneParamFunction extends ScalarFunction[Int] with
ReducibleFunction[Int, Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "zero_or_one"
+ override def canonicalName(): String = name()
+ override def toString: String = name()
+ override def produceResult(input: InternalRow): Int = input.getInt(0)
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[Int, Int] = {
+ if (otherFunc == ZeroOrOneParamFunction && thisParams.length !=
otherParams.length) {
+ BucketReducer(1)
+ } else {
+ null
+ }
+ }
+}
+
+/**
+ * A transform with a `CalendarIntervalType` literal parameter (which is
non-complex but not an
+ * `AtomicType`). Used to verify such a parameter is not rejected as a complex
container before its
+ * reducer is consulted.
+ */
+object IntervalParamFunction extends ScalarFunction[Int] with
ReducibleFunction[Int, Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType,
CalendarIntervalType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "interval_param"
+ override def canonicalName(): String = name()
+ override def toString: String = name()
+ override def produceResult(input: InternalRow): Int = input.getInt(0)
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[Int, Int] = {
+ if (otherFunc == IntervalParamFunction) BucketReducer(1) else null
+ }
+}
+
+/** A user type whose UDT serializes to a struct (an [[InternalRow]]). */
+class StructBacked(val n: Int) extends Serializable
+
+/**
+ * A UDT whose `sqlType` is a [[StructType]]: a literal of this type carries
an [[InternalRow]]
+ * value, yet its `dataType` is the UDT, not `StructType`. This is the case a
`DataType`-based
+ * container check misses but a value-based one catches.
+ */
+class StructBackedUDT extends UserDefinedType[StructBacked] {
+ override def sqlType: DataType = StructType(Seq(StructField("n",
IntegerType, nullable = false)))
+ override def serialize(obj: StructBacked): InternalRow = new
GenericInternalRow(Array[Any](obj.n))
+ override def deserialize(datum: Any): StructBacked = datum match {
+ case row: InternalRow => new StructBacked(row.getInt(0))
+ }
+ override def userClass: Class[StructBacked] = classOf[StructBacked]
+}
+
+/**
+ * A transform whose declared input type at the literal position is a UDT
([[StructBackedUDT]]).
+ * Used to make the value-based `noComplexLiteralParams` guard load-bearing: a
UDT-over-struct
+ * literal matches the declared input type (so `literalParamsMatchInputTypes`
passes), yet its value
+ * is an [[InternalRow]], so only the value-based guard can reject it. The
reducer returns
+ * unconditionally, so reaching it at all is the leak.
+ */
+object UdtParamFunction extends ScalarFunction[Int] with
ReducibleFunction[Int, Int] {
+ override def inputTypes(): Array[DataType] = Array(new StructBackedUDT,
LongType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "udt_param"
+ override def canonicalName(): String = name()
+ override def toString: String = name()
+ override def produceResult(input: InternalRow): Int = input.getInt(1)
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[Int, Int] = BucketReducer(1)
+}
+
+/**
+ * A string truncate whose reducer reads its width type-tolerantly (via
[[Number]], so it accepts a
+ * boxed Short or Integer). Used to verify that the literal-param-type gate --
not an incidental
+ * ClassCastException in the connector -- is what makes a mismatched-type
(e.g. ShortType) width
+ * non-reducible. With the gate removed, this reducer WOULD reduce a
ShortType-width pair.
+ */
+object TypeTolerantTruncateFunction
+ extends ScalarFunction[UTF8String]
+ with ReducibleFunction[UTF8String, UTF8String] {
+ override def inputTypes(): Array[DataType] = Array(StringType, IntegerType)
+ override def resultType(): DataType = StringType
+ override def name(): String = "tolerant_truncate"
+ override def canonicalName(): String = name()
+ override def toString: String = name()
+ override def produceResult(input: InternalRow): UTF8String =
+ input.getUTF8String(0).substring(0, input.getInt(1))
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[UTF8String, UTF8String] = {
+ if (otherFunc == TypeTolerantTruncateFunction &&
+ thisParams.length == 1 && otherParams.length == 1) {
+ val thisWidth = thisParams(0).value().asInstanceOf[Number].intValue()
+ val otherWidth = otherParams(0).value().asInstanceOf[Number].intValue()
+ val smaller = math.min(thisWidth, otherWidth)
+ if (smaller != thisWidth) return TruncateReducer(smaller)
+ }
+ null
+ }
+}
+
+/**
+ * A function whose generalized reducer throws an unexpected
(non-UnsupportedOperationException)
+ * exception. Used to verify the dispatch logs it and treats the pair as not
reducible (a shuffle),
+ * rather than crashing or emitting the misleading "implements no reducer"
hint (it does implement
+ * the overload -- it threw a bug, which is a different signal than
UOE-means-unimplemented).
+ */
+object ThrowingReducerFunction extends ScalarFunction[Int] with
ReducibleFunction[Int, Int] {
+ override def inputTypes(): Array[DataType] = Array(IntegerType, IntegerType)
+ override def resultType(): DataType = IntegerType
+ override def name(): String = "throwing_reducer"
+ override def canonicalName(): String = name()
+ override def toString: String = name()
+ override def produceResult(input: InternalRow): Int = input.getInt(1)
+
+ override def reducer(
+ thisParams: Array[Literal[_]],
+ otherFunc: ReducibleFunction[_, _],
+ otherParams: Array[Literal[_]]): Reducer[Int, Int] =
+ throw new RuntimeException("boom from reducer")
+}
+
+/**
+ * A bucket-like function implementing BOTH reducer overloads: the deprecated
int overload succeeds
+ * (returns a reducer), while the generalized Literal[] overload throws. Used
to verify the
+ * single-int dispatch is lazy -- it must not invoke (and log the throw from)
the generalized
Review Comment:
**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.
--
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]
