dongjoon-hyun commented on code in PR #60078:
URL: https://github.com/apache/spark/pull/60078#discussion_r4464418876
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -39,6 +39,10 @@ class InternalRowComparableWrapper private (
@transient private var _structType: StructType,
@transient private var _ordering: BaseOrdering) extends Serializable {
+ require(row.numFields == dataTypes.length,
+ s"The row has ${row.numFields} fields, but the types it is compared at
have " +
+ s"${dataTypes.length}")
Review Comment:
This `require` looks like a behavior change rather than a pure plan-string
improvement. Previously, a connector whose `partitionKey()` row is wider than
the reported `KeyGroupedPartitioning` expressions (e.g., a row `(a, b)` with
keys `[a]`) kept working, because the ordering, the Murmur3 hash and
`KeyGroupedPartitioner` only read the first `n` fields. With this PR, the same
query fails at planning in `DataSourceV2ScanExecBase.reportedKeyedPartitioning`
-> `KeyedPartitioning.apply` with `IllegalArgumentException: requirement
failed: ...`, without an error condition or any hint about the connector/scan.
The same applies to external callers of the public `@deprecated` constructor.
The PR description says `No query result or execution behavior changes`.
Could you either move this validation to the scan boundary with a proper error
condition (like the existing `SparkException` checks in `PushDownUtils`), or
document the behavior change in the PR description and the migration guide?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -39,6 +39,10 @@ class InternalRowComparableWrapper private (
@transient private var _structType: StructType,
@transient private var _ordering: BaseOrdering) extends Serializable {
+ require(row.numFields == dataTypes.length,
Review Comment:
For the narrower-row case, this check is not reached.
`DataSourceV2ScanExecBase.reportedKeyedPartitioning` sorts the raw
`partitionKey()` rows with `groupedKeyRowOrdering` before wrapping them:
```scala
val keys =
inputPartitions.map(_.asInstanceOf[HasPartitionKey].partitionKey())
.sorted(KeyedPartitioning.groupedKeyRowOrdering(exprs.map(_.dataType)))
Some(KeyedPartitioning(exprs, keys))
```
If the rows have 1 field for 2 expressions and two splits share the same
first field, the comparison reads field 1, which throws
`ArrayIndexOutOfBoundsException` for `GenericInternalRow` (or reads beyond the
row for `UnsafeRow` when assertions are disabled). So the clear error message
only covers wider rows. Validating the width before sorting would cover both
directions.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -65,6 +69,9 @@ class InternalRowComparableWrapper private (
_ordering
}
+ override def toString: String =
+ row.toSeq(dataTypes).mkString("[", ",", "]")
Review Comment:
`row.toSeq(dataTypes)` returns Catalyst internal values, so the plan string
shows the internal representation for several common partition key types:
- `DateType` (e.g., identity partitioning on a `DATE` column): epoch days
such as `[19723]`
- `TimestampType` / `TimestampNTZType`: microseconds as a `Long`
- `TimeType` and interval types: raw numbers
- `BinaryType`: `[B@6d06d69c`, which is still the object identity this PR
intends to remove, and it differs from run to run.
How about formatting each value with `Literal(value, dataType).toString`,
which already handles these types?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -39,6 +39,10 @@ class InternalRowComparableWrapper private (
@transient private var _structType: StructType,
@transient private var _ordering: BaseOrdering) extends Serializable {
+ require(row.numFields == dataTypes.length,
Review Comment:
This `require` also runs once per input row in the shuffle:
`ShuffleExchangeExec.getPartitionKeyExtractor` calls
`wrapperFactory(partitionKeyRow)` for every record. There, `partitionKeyRow` is
created with `expressions.length` fields, so the check is always true by
construction, and `dataTypes.length` is O(n) when `dataTypes` is a `List`.
Checking once at the connector boundary (or in `KeyedPartitioning.apply`) would
avoid the per-row cost.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -750,15 +750,30 @@ case class KeyedPartitioning(
/**
* Prints the layout's contents where the value object would print, which
keeps the plan string as
* it was before the layout held them. The list is curated rather than the
layout's own fields,
- * for two reasons. `partitionKeys` has to be printed as a `Seq` for
`maxFields` to truncate it,
- * and a partitioning can hold one key per split. And `dataTypes` has its
naming erased, so
- * printing it would put a struct field name into a plan that appears
nowhere in the query. A
- * field the layout grows is therefore a decision here, and the default is
to print it: the plan
- * string is the only place a reader sees why a marked partitioning still
shuffles.
+ * for two reasons. A partitioning can hold one key per split, so
`partitionKeys` is summarized:
+ * as many keys as `spark.sql.debug.maxToStringFields` has room for, then
their total. And
+ * `dataTypes` has its naming erased, so printing it would put a struct
field name into a plan
+ * that appears nowhere in the query. A field the layout grows is therefore
a decision here, and
+ * the default is to print it: the plan string is the only place a reader
sees why a marked
+ * partitioning still shuffles.
*/
- override protected def stringArgs: Iterator[Any] =
- Iterator(expressions, partitionKeys, isGrouped, isCollapsed) ++
unknownKeysOrigin ++
- ungroupingOrigin
+ override protected def stringArgs: Iterator[Any] = {
+ // `Expression.toString` truncates the flattened argument list at
`maxToStringFields`. The
+ // expressions and the fields printed after the keys keep their slots, and
the keys take what
+ // is left; when they cannot all fit, their total takes one key slot. When
there is not even a
+ // slot left for that, the total still wins over a trailing field, which
the outer truncation
+ // then drops.
+ val trailing = Seq[Any](isGrouped, isCollapsed) ++ unknownKeysOrigin ++
ungroupingOrigin
+ val keyBudget = math.max(0, SQLConf.get.maxToStringFields -
expressions.size - trailing.size)
+ val keys: Seq[Any] =
+ if (partitionKeys.length <= keyBudget) {
+ partitionKeys
+ } else {
+ partitionKeys.take(math.max(0, keyBudget - 1)) :+
Review Comment:
When `keyBudget` is 0, the total wins over the trailing fields, and the
outer truncation drops them from the end. So `unknownKeysOrigin` /
`ungroupingOrigin` are the first to go, although the scaladoc above says the
plan string is the only place where a reader sees why a marked partitioning
still shuffles. It also produces two ellipses in one string, as the new test
shows:
```
keyedpartitioning(x, ... 12 partition keys in total, true, ... 1 more fields)
```
Would it be better to keep the trailing fields and drop the key summary
first in this case?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala:
##########
@@ -65,6 +69,9 @@ class InternalRowComparableWrapper private (
_ordering
}
+ override def toString: String =
Review Comment:
Minor: `equals`/`hashCode` are collation-aware, but `toString` is not. For
example, with `UTF8_LCASE`, `'a'` and `'A'` are the same key for Spark, but
they print as `[a]` and `[A]`. This can make duplicate (ungrouped) keys, or the
two sides of a paired join, look like different keys while debugging. Not a
blocker, but it might be worth a note in the scaladoc.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -750,15 +750,30 @@ case class KeyedPartitioning(
/**
* Prints the layout's contents where the value object would print, which
keeps the plan string as
* it was before the layout held them. The list is curated rather than the
layout's own fields,
- * for two reasons. `partitionKeys` has to be printed as a `Seq` for
`maxFields` to truncate it,
- * and a partitioning can hold one key per split. And `dataTypes` has its
naming erased, so
- * printing it would put a struct field name into a plan that appears
nowhere in the query. A
- * field the layout grows is therefore a decision here, and the default is
to print it: the plan
- * string is the only place a reader sees why a marked partitioning still
shuffles.
+ * for two reasons. A partitioning can hold one key per split, so
`partitionKeys` is summarized:
+ * as many keys as `spark.sql.debug.maxToStringFields` has room for, then
their total. And
+ * `dataTypes` has its naming erased, so printing it would put a struct
field name into a plan
+ * that appears nowhere in the query. A field the layout grows is therefore
a decision here, and
+ * the default is to print it: the plan string is the only place a reader
sees why a marked
+ * partitioning still shuffles.
*/
- override protected def stringArgs: Iterator[Any] =
- Iterator(expressions, partitionKeys, isGrouped, isCollapsed) ++
unknownKeysOrigin ++
- ungroupingOrigin
+ override protected def stringArgs: Iterator[Any] = {
+ // `Expression.toString` truncates the flattened argument list at
`maxToStringFields`. The
+ // expressions and the fields printed after the keys keep their slots, and
the keys take what
+ // is left; when they cannot all fit, their total takes one key slot. When
there is not even a
+ // slot left for that, the total still wins over a trailing field, which
the outer truncation
+ // then drops.
+ val trailing = Seq[Any](isGrouped, isCollapsed) ++ unknownKeysOrigin ++
ungroupingOrigin
+ val keyBudget = math.max(0, SQLConf.get.maxToStringFields -
expressions.size - trailing.size)
+ val keys: Seq[Any] =
+ if (partitionKeys.length <= keyBudget) {
Review Comment:
Nit: `KeyLayout.partitionKeys` is `@transient` (`Driver-side only`), so it
is `null` in a deserialized `KeyedPartitioning`. The previous `stringArgs`
printed `null` in that case, but `partitionKeys.length` / `take` here would
throw `NullPointerException`. If a deserialized instance is ever stringified
(e.g., in a log or an error message on an executor), the NPE would hide the
original error. A null guard would be cheap.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -750,15 +750,30 @@ case class KeyedPartitioning(
/**
* Prints the layout's contents where the value object would print, which
keeps the plan string as
* it was before the layout held them. The list is curated rather than the
layout's own fields,
- * for two reasons. `partitionKeys` has to be printed as a `Seq` for
`maxFields` to truncate it,
- * and a partitioning can hold one key per split. And `dataTypes` has its
naming erased, so
- * printing it would put a struct field name into a plan that appears
nowhere in the query. A
- * field the layout grows is therefore a decision here, and the default is
to print it: the plan
- * string is the only place a reader sees why a marked partitioning still
shuffles.
+ * for two reasons. A partitioning can hold one key per split, so
`partitionKeys` is summarized:
+ * as many keys as `spark.sql.debug.maxToStringFields` has room for, then
their total. And
+ * `dataTypes` has its naming erased, so printing it would put a struct
field name into a plan
+ * that appears nowhere in the query. A field the layout grows is therefore
a decision here, and
+ * the default is to print it: the plan string is the only place a reader
sees why a marked
+ * partitioning still shuffles.
*/
- override protected def stringArgs: Iterator[Any] =
- Iterator(expressions, partitionKeys, isGrouped, isCollapsed) ++
unknownKeysOrigin ++
- ungroupingOrigin
+ override protected def stringArgs: Iterator[Any] = {
+ // `Expression.toString` truncates the flattened argument list at
`maxToStringFields`. The
+ // expressions and the fields printed after the keys keep their slots, and
the keys take what
+ // is left; when they cannot all fit, their total takes one key slot. When
there is not even a
+ // slot left for that, the total still wins over a trailing field, which
the outer truncation
+ // then drops.
+ val trailing = Seq[Any](isGrouped, isCollapsed) ++ unknownKeysOrigin ++
ungroupingOrigin
+ val keyBudget = math.max(0, SQLConf.get.maxToStringFields -
expressions.size - trailing.size)
+ val keys: Seq[Any] =
+ if (partitionKeys.length <= keyBudget) {
+ partitionKeys
+ } else {
+ partitionKeys.take(math.max(0, keyBudget - 1)) :+
Review Comment:
Nit: `partitionKeys.length` is the same as the existing `numPartitions` val,
and `math.max(0, keyBudget - 1)` is not needed because `take` returns an empty
`Seq` for a negative count. `partitionKeys.take(keyBudget - 1) :+ s"...
$numPartitions partition keys in total"` would be simpler.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/DistributionSuite.scala:
##########
@@ -508,6 +508,49 @@ class DistributionSuite extends SparkFunSuite with
SQLHelper {
assert(interned.layout eq kpX.layout, "the whole layout is interned, keys
and all")
}
+ test("SPARK-60859: a KeyedPartitioning plan string shows the keys that fit
and their total") {
+ val x = AttributeReference("x", IntegerType)()
+ val keys = (1 to 12).map(InternalRow(_))
+
+ // One expression, the total and the two flags take four of the ten slots.
+ withSQLConf(SQLConf.MAX_TO_STRING_FIELDS.key -> "10") {
+ assert(KeyedPartitioning(Seq(x), keys).toString ==
+ s"keyedpartitioning($x, [1], [2], [3], [4], [5], [6], " +
+ "... 12 partition keys in total, true, false)")
+
+ // When every key fits, the total is not needed.
+ assert(KeyedPartitioning(Seq(x), keys.take(6)).toString ==
Review Comment:
Could you add the boundary case where the keys fill the budget exactly (7
keys with `maxToStringFields = 10`), and an empty `partitionKeys` case?
Currently, changing `<=` to `<` in `partitionKeys.length <= keyBudget` would
still pass all assertions.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -750,15 +750,30 @@ case class KeyedPartitioning(
/**
* Prints the layout's contents where the value object would print, which
keeps the plan string as
Review Comment:
Nit: `which keeps the plan string as it was before the layout held them` now
reads stale, since this PR changes the plan string (the keys are printed as
values and summarized with their total).
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapperSuite.scala:
##########
@@ -153,4 +154,24 @@ class InternalRowComparableWrapperSuite extends
SparkFunSuite {
assert(wrap(IntegerType, InternalRow(1)) !== wrap(LongType,
InternalRow(1L)),
"and a type that decides where a value belongs still tells them apart")
}
+
+ test("SPARK-60859: a wrapper rejects a row whose width disagrees with its
types") {
+ val factory = InternalRowComparableWrapper
+ .getInternalRowComparableWrapperFactory(Seq(IntegerType, IntegerType))
+ val e = intercept[IllegalArgumentException] {
+ factory(InternalRow(1, 2, 3))
+ }
+ assert(e.getMessage.contains("The row has 3 fields, but the types it is
compared at have 2"))
+ }
+
+ test("SPARK-60859: a wrapper prints its row values") {
+ val factory = InternalRowComparableWrapper
+ .getInternalRowComparableWrapperFactory(Seq(IntegerType, StringType))
+ assert(factory(InternalRow(1, UTF8String.fromString("a"))).toString ===
"[1,a]")
+
+ // An UnsafeRow has no toString of its own, so the wrapper has to read the
row.
Review Comment:
`UnsafeRow` does have its own `toString` (`UnsafeRow.java`), which prints a
hex dump of its 8-byte words instead of the values. I guess the point is that
it does not print the values, so the wrapper has to read them with the types.
Could you update this comment (and the PR description) accordingly?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala:
##########
@@ -750,15 +750,30 @@ case class KeyedPartitioning(
/**
* Prints the layout's contents where the value object would print, which
keeps the plan string as
* it was before the layout held them. The list is curated rather than the
layout's own fields,
- * for two reasons. `partitionKeys` has to be printed as a `Seq` for
`maxFields` to truncate it,
- * and a partitioning can hold one key per split. And `dataTypes` has its
naming erased, so
- * printing it would put a struct field name into a plan that appears
nowhere in the query. A
- * field the layout grows is therefore a decision here, and the default is
to print it: the plan
- * string is the only place a reader sees why a marked partitioning still
shuffles.
+ * for two reasons. A partitioning can hold one key per split, so
`partitionKeys` is summarized:
+ * as many keys as `spark.sql.debug.maxToStringFields` has room for, then
their total. And
+ * `dataTypes` has its naming erased, so printing it would put a struct
field name into a plan
+ * that appears nowhere in the query. A field the layout grows is therefore
a decision here, and
+ * the default is to print it: the plan string is the only place a reader
sees why a marked
+ * partitioning still shuffles.
*/
- override protected def stringArgs: Iterator[Any] =
- Iterator(expressions, partitionKeys, isGrouped, isCollapsed) ++
unknownKeysOrigin ++
- ungroupingOrigin
+ override protected def stringArgs: Iterator[Any] = {
+ // `Expression.toString` truncates the flattened argument list at
`maxToStringFields`. The
+ // expressions and the fields printed after the keys keep their slots, and
the keys take what
+ // is left; when they cannot all fit, their total takes one key slot. When
there is not even a
+ // slot left for that, the total still wins over a trailing field, which
the outer truncation
+ // then drops.
+ val trailing = Seq[Any](isGrouped, isCollapsed) ++ unknownKeysOrigin ++
ungroupingOrigin
+ val keyBudget = math.max(0, SQLConf.get.maxToStringFields -
expressions.size - trailing.size)
Review Comment:
This budget arithmetic re-implements how `Expression.toString` flattens and
truncates its arguments (it reads `SQLConf.get.maxToStringFields` separately
and assumes one slot per flattened element), so it can silently drift if
`flatArguments` or the truncation changes. An alternative is to emit the keys
as a single pre-truncated argument, e.g., `truncatedString(partitionKeys, "[",
", ", "]", maxFields)`, or to override `toString`/`simpleString`. Either way,
there would be no slot math and no double ellipsis.
--
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]
