dongjoon-hyun commented on code in PR #58523:
URL: https://github.com/apache/spark/pull/58523#discussion_r3935169708
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -880,9 +880,13 @@ object KeyedPartitioning {
* `PartitioningCollection`, whose invariant requires equal partition keys
-- but join types that
* expose only one side's partitioning (e.g. LEFT OUTER) run nothing that
compares the two
* orders, and silently return wrong results.
+ *
+ * It is the keys' own ordering, the one
`InternalRowComparableWrapper.equals` compares with, so
+ * one definition answers both. `EnsureRequirements`' `OrderedDistribution`
arm is the one place
+ * that lays grouped keys out in another order, the distribution's own.
*/
def groupedKeyRowOrdering(dataTypes: Seq[DataType]): BaseOrdering =
- RowOrdering.createNaturalAscendingOrdering(dataTypes)
+ InternalRowComparableWrapper.orderingFor(dataTypes)
Review Comment:
The caveat in the description is accurate, and I checked that nothing in the
current suites hits it: every `SharedSparkSession` suite runs under
`CODEGEN_ONLY` with fallback off, `NO_CODEGEN` appears only inside
`withSQLConf` blocks that never build a `KeyedPartitioning`, and no other
construction-time conf affects comparison results (collation is already part of
`StringType`). Two things it does change for these four sites, which used to
re-read the mode per call: the first loader in the JVM decides codegen vs
interpreted for the rest of the JVM's life, and in `FALLBACK` mode a transient
codegen failure pins the interpreted instance permanently. If you'd rather keep
per-mode fidelity, keying the cache on `(SQLConf.get.codegenFactoryMode,
dataTypes)` in the load path is cheap and keeps the identity the new test
asserts within any one mode. Accepting the caveat as written is also defensible.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -115,6 +115,9 @@ object InternalRowComparableWrapper {
new InternalRowComparableWrapper(partitionRow,
partitionExpression.map(_.dataType))
}
+ /** The cached ordering a wrapper of these `dataTypes` compares its rows
with in `equals`. */
+ def orderingFor(dataTypes: Seq[DataType]): BaseOrdering =
orderingCache.get(dataTypes)
Review Comment:
One pre-existing property of this cache that the PR extends to the
sort/group sites: the key is `Seq[DataType]`, but
`PythonUserDefinedType.equals`/`hashCode` compare only `pyUDT` and ignore
`sqlType` (`sql/api/.../UserDefinedType.scala`). `supportsExpressions` does not
gate UDTs out and `OrderUtils.isOrderable` accepts them via `sqlType`, so two
Python UDTs with the same class name but different `sqlType` (e.g. two Connect
clients on different versions of the same class) share one entry, and the
second gets an ordering built for the first's `sqlType`. Before this PR the
sort sites built a fresh ordering from the exact types passed; now they can get
the mismatched one. Very much an edge case and the root cause is UDT equality,
not the cache, but worth knowing that `groupedKeyRowOrdering` now inherits it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -92,8 +92,7 @@ trait DataSourceV2ScanExecBase
keyGroupedPartitioning match {
case Some(exprs) if conf.v2BucketingEnabled &&
KeyedPartitioning.supportsExpressions(exprs) &&
inputPartitions.nonEmpty &&
inputPartitions.forall(_.isInstanceOf[HasPartitionKey]) =>
- val dataTypes = exprs.map(_.dataType)
- val rowOrdering = RowOrdering.createNaturalAscendingOrdering(dataTypes)
+ val rowOrdering =
KeyedPartitioning.groupedKeyRowOrdering(exprs.map(_.dataType))
val partitionKeys =
inputPartitions.map(_.asInstanceOf[HasPartitionKey].partitionKey()).sorted(rowOrdering)
KeyedPartitioning(exprs, partitionKeys)
Review Comment:
Minor: `KeyedPartitioning.apply` recomputes `expressions.map(_.dataType)`
and its factory does a second `orderingCache.get` for the same key, so per call
this path builds the type list twice and takes the key lock three times. Since
`apply` does not need pre-sorted input (`isGrouped` is order-independent),
building the partitioning first and sorting with its own `keyOrdering`, the way
`PushDownUtils` already does with `keyRowOrdering`, would drop the duplicate.
Micro-level next to the sort itself, so only if you are touching this anyway.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2ScanExecBase.scala:
##########
@@ -92,8 +92,7 @@ trait DataSourceV2ScanExecBase
keyGroupedPartitioning match {
Review Comment:
The 27-calls-per-scan count that motivates this PR comes from
`outputPartitioning` being a `def`. After this change each call still maps
every input partition to its key, sorts them, and then
`KeyedPartitioning.apply` wraps each key and runs a Murmur3-hashed `.distinct`,
so for scans with hundreds or thousands of partitions the per-call work is
still well above the ~100us removed here. Making it `@transient override lazy
val outputPartitioning` looks safe: all four concrete scans define
`inputPartitions` as a `lazy val`, `keyGroupedPartitioning` is a constructor
parameter, `doCanonicalize` goes through `copy`, and
`BatchScanExec.filteredPartitions` consumes the pre-filter partitioning, which
is exactly the cached value. `FileSourceScanExec` already does this (`lazy val
(outputPartitioning, outputOrdering)`). Fine as a follow-up if you'd rather
keep this PR minimal, but it would subsume most of the benefit claimed here.
--
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]
