dongjoon-hyun commented on code in PR #59165:
URL: https://github.com/apache/spark/pull/59165#discussion_r4168744978
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -2086,6 +2119,48 @@ case class KeyedShuffleSpec(
}
}
+ /**
+ * Whether every argument of `transform` is a bare column.
`isExpressionCompatible` asks it of
+ * both sides before it compares anything, `canCreatePartitioning` asks it
of a template, and
+ * `rebuiltOver` asserts it.
+ *
+ * `keyPositions` pairs two sides up by the column each expression
references. That pairing
+ * describes one key space when the argument is the column itself.
`bucket(4, b + 1)` is a
+ * function of `b`, but not the same function of it as `bucket(4, x)` is of
`x`. Nor is
+ * `bucket(4, s.a)` the same function of `s` as `bucket(4, t.b)` is of `t`.
So such a transform
+ * pairs only with a transform over the same `argumentShape`. Over the same
shape, the two
+ * compare as two transforms of columns do, e.g. `bucket(4, b + 1)` with
`bucket(4, c + 1)` on a
+ * join `b = c`. A reduce of `bucket(8, b + 1)` onto `bucket(4, c + 1)` is
fine too, since it
+ * maps the partition keys and evaluates no argument. An identity side is
not reduced onto such a
+ * transform, since that would evaluate `id + 1` on keys the query never
computes it for, which
+ * can overflow.
+ */
+ private def argumentsAreColumns(transform: TransformExpression): Boolean =
+ transform.children.forall(_.isInstanceOf[Attribute])
+
+ /**
+ * `transform` with its argument replaced by `column`. Only a transform of
bare columns can be
+ * rebuilt this way. Any other argument would lose what surrounds its
column, e.g. the `+ 1` of
+ * `bucket(4, b + 1)`.
+ */
+ private def rebuiltOver(transform: TransformExpression, column: Expression)
Review Comment:
Nit: the parameter name says that the replacement is a column, but only the
source has to be one, which is what the assert checks. `createPartitioning`
passes the other child's join key at L2298, and that is an expression in
exactly the cases this PR handles: `bucket(4, t1.id)` becomes `bucket(4, p.b +
1)`, as the same-shape tests in `KeyGroupedPartitioningSuite` expect. Could it
be named `key`, with the doc saying "`transform`, whose arguments are bare
columns, with each replaced by `key`, which may be any expression"?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -1941,8 +1949,20 @@ case class KeyedShuffleSpec(
* `KeyedPartitioning.positionsCoveringClusterKeys` counts it. See that
method for why the two
* stay apart.
*
- * The assertion is what makes the reference reading sound. With one
reference per expression,
- * "some reference is a cluster key" and "every reference is" are the same
statement.
+ * Only an expression with a single reference maps to a position. With one
reference, "some
+ * reference is a cluster key" and "every reference is" are the same
statement, which is what
+ * makes the reference reading sound. A scan reports no other expression
+ * (`KeyedPartitioning.supportsExpressions`), but two planner paths build
one over the other
Review Comment:
Minor: a scan cannot report a transform of a struct field either, so the
cost that the PR description lists for "a scan that reports a bucket of a
struct field, e.g. `bucket(4, s.a)`" does not seem to arise.
`V2ExpressionUtils.resolveRef` resolves `s.a` through `LogicalPlan.resolve`,
which wraps a nested field in an `Alias` (package.scala:405-410). So the scan's
transform is over an `Alias`, which `supportsExpressions` (L1110-L1117)
refuses, since it accepts only an `Attribute` or a `GetStructField` chain.
`DataSourceV2ScanExecBase` then reports no `KeyedPartitioning`
(DataSourceV2ScanExecBase.scala:105), and I found nothing in between that
strips the alias. Such a table gets no storage-partitioned join on master
either, which would also explain why the in-memory catalog cannot test it.
Could the description drop that bullet, or say that such layouts come only from
joins? I traced this but did not run it.
--
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]
