ulysses-you commented on code in PR #60078:
URL: https://github.com/apache/spark/pull/60078#discussion_r4465669186
##########
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:
Took the second alternative: toString/simpleString are overridden, and both
the key list and the argument list are rendered with truncatedString. There is
no manual flattening of the argument list.
##########
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:
Guarded: a deserialized partitioning prints null. Covered by a
JavaSerializer round-trip test.
##########
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:
The summary now goes through truncatedString, so the take/math.max
arithmetic is gone.
##########
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:
Rewritten to describe the current fields and the collation caveat.
##########
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:
Added: exact fit (7 keys at maxFields = 10), 8 keys (truncated), an empty
key list, and the deserialized null case.
##########
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:
Fixed: UnsafeRow's own toString is a hex dump of its words; the test comment
now says so.
--
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]
