peter-toth commented on code in PR #58501:
URL: https://github.com/apache/spark/pull/58501#discussion_r3944837325
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -582,53 +590,55 @@ case class CoalescedNullAwareHashPartitioning(
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
+ keyDataTypes: Seq[DataType],
isGrouped: Boolean,
isCollapsed: Boolean) extends Expression with Partitioning with
Unevaluable {
override val numPartitions = partitionKeys.length
+ // The keys carry their own types, so the field can be checked against the
thing it describes
+ // rather than argued about. One key is enough: `concat` is the only copy
that mixes rows from
+ // several partitionings, and it checks all of them.
+ require(keyDataTypes.length == expressions.length,
+ "A KeyedPartitioning must have one key data type per partition expression")
+ require(partitionKeys.headOption.forall(_.dataTypes == keyDataTypes),
+ "A KeyedPartitioning's keyDataTypes must be the types its partitionKeys
were built with")
+
override def children: Seq[Expression] = expressions
override def nullable: Boolean = false
override def dataType: DataType = IntegerType
+ /**
+ * Drops the `keyDataTypes`, so that `explain` shows what it showed before
the field existed. They
+ * are the types of the keys printed beside them, which adds nothing a
reader of a plan wants.
+ */
+ override protected def stringArgs: Iterator[Any] =
Review Comment:
Dropped, along with the field it was hiding. `stringArgs` is master's again.
Your point stands for SPARK-59285, and it lands the other way there. The
types are erased, so printing them would put a struct field named `0` into
every explain. The override is kept there with a comment saying that, rather
than arguing the information is worthless.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -582,53 +590,55 @@ case class CoalescedNullAwareHashPartitioning(
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
+ keyDataTypes: Seq[DataType],
isGrouped: Boolean,
isCollapsed: Boolean) extends Expression with Partitioning with
Unevaluable {
override val numPartitions = partitionKeys.length
+ // The keys carry their own types, so the field can be checked against the
thing it describes
+ // rather than argued about. One key is enough: `concat` is the only copy
that mixes rows from
+ // several partitionings, and it checks all of them.
+ require(keyDataTypes.length == expressions.length,
+ "A KeyedPartitioning must have one key data type per partition expression")
+ require(partitionKeys.headOption.forall(_.dataTypes == keyDataTypes),
+ "A KeyedPartitioning's keyDataTypes must be the types its partitionKeys
were built with")
+
override def children: Seq[Expression] = expressions
override def nullable: Boolean = false
override def dataType: DataType = IntegerType
+ /**
+ * Drops the `keyDataTypes`, so that `explain` shows what it showed before
the field existed. They
+ * are the types of the keys printed beside them, which adds nothing a
reader of a plan wants.
+ */
+ override protected def stringArgs: Iterator[Any] =
+ Iterator(expressions, partitionKeys, isGrouped, isCollapsed)
+
override protected def withNewChildrenInternal(
newChildren: IndexedSeq[Expression]): KeyedPartitioning =
copy(expressions = newChildren)
- /** Need not be what the `partitionKeys` rows hold. See `keyDataTypes`. */
- @transient lazy val expressionDataTypes: Seq[DataType] =
expressions.map(_.dataType)
-
/**
- * The types the `partitionKeys` rows were built with. Anything reading
those rows should take its
- * types from here. It is a driver-side value, since `partitionKeys` is
`@transient`.
- *
- * They differ from the `expressionDataTypes` in two cases. A join that
reduced both sides' keys
- * onto a key space no transform names leaves a marked expression whose type
can be anything, see
- * `expressionsDescribeKeys`. A one-side reduce keeps them equal, because
the expression the
- * partitioning then reports is the target transform and
`EnsureRequirements` refuses a reducer
- * whose result type disagrees with it.
`KeyedShuffleSpec.createPartitioning` is the other case.
- * It puts the other child's expressions over these keys with no reducer in
sight, so a struct
- * field can be named differently on the two sides. With no key at all the
expressions are all
- * there is, and there is no row to read or to place.
+ * The types the partition expressions produce. Not what the `partitionKeys`
rows hold, whenever
+ * the expressions have stopped describing the keys, which happens in two
ways.
Review Comment:
Taken, and your (b) changed shape rather than going away.
`KeyedShuffleSpec.createPartitioning` re-targeting the expressions no longer
makes the two answers differ, because the erasure brings both to one answer. So
the doc now lists the naming, since `expressionDataTypes` is not erased at all,
and the both-sides reduce. It also says plainly that a one-side reduce keeps
the two equal up to the naming, because the expression the partitioning reports
is the target transform and `EnsureRequirements` refuses a reducer whose result
type disagrees with it.
Your underlying point is what I got wrong: `expressionsDescribeKeys` only
covers the reduce case. The `ShuffleExchangeExec` paragraph now says that, and
the comment at the partitioner build site gives the reason the re-wrap earns
its keep, rather than the struct-field-name one it used to give.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -582,53 +590,55 @@ case class CoalescedNullAwareHashPartitioning(
case class KeyedPartitioning(
expressions: Seq[Expression],
@transient partitionKeys: Seq[InternalRowComparableWrapper],
+ keyDataTypes: Seq[DataType],
Review Comment:
Taken, and it is the deeper fix. I owe you two things on it.
First, I tried `TypedKeys` and it did not settle the pairing. A value object
per partitioning still lets two members of a `PartitioningCollection` hold two
different pairs. What settles it is `KeyLayout(partitionKeys, dataTypes,
isGrouped, isCollapsed)` shared **by reference**, so the collection's invariant
is one `eq` plus the arity, `createPartitioning` has nothing to decide, and
`PartitionGrouping` becomes the layout it will report. That is SPARK-59285, and
the attempt is kept as history.
Second, `projectKeys` and `reduceKeys` still return tuples in this PR, and
the `fold` seeds in `EnsureRequirements` still pair them by hand. My own
cleanup pass flagged the same thing independently, so I am not going to argue
it is fine. It is deliberately left for that ticket to keep this one
backportable.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -572,6 +572,14 @@ case class CoalescedNullAwareHashPartitioning(
* comparison and grouping. One per partition. Typically
in sorted order when
* produced by a data source or `GroupPartitionsExec`,
but this is not
* guaranteed after projection. May contain duplicates
when ungrouped.
+ * @param keyDataTypes The types the `partitionKeys` rows were built with, one
per expression.
Review Comment:
Taken. Those inventories went with the field.
`keyDataTypes`' doc no longer lists copy sites, and `concat` has no
`keyDataTypes` clause to document. What is left there is the contract plus the
one thing a reader cannot derive: with no key row the expressions answer, and
after a both-sides reduce that answer is a type no key of the partitioning
holds.
--
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]
