spark-test commented on code in PR #60078: URL: https://github.com/apache/spark/pull/60078#discussion_r4469640747
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala: ########## @@ -748,17 +748,32 @@ case class KeyedPartitioning( override def dataType: DataType = IntegerType /** - * 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. + * Prints the layout as the partition keys followed by the flags and origins, with the keys + * summarized when there are more than `maxFields` of them. `dataTypes` is left out because its Review Comment: **Nit (P3):** This says the keys are summarized when there are more than `maxFields` of them, but `simpleString` summarizes them once they exceed `keyBudget = maxFields - expressions.size - trailing.size`. With `maxToStringFields = 10` and one expression, 8 keys are already summarized as `[[1], [2], [3], [4], [5], [6], [7], ... 1 more fields]`, which the new `DistributionSuite` test asserts. At the default of 25, a one-expression partitioning is summarized past 22 keys, past 21 with one origin, and past 20 with both. This extends an [existing discussion](https://github.com/apache/spark/pull/60078#discussion_r4464418944). See **Shared repair plan 2** in the review body. ########## sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapperSuite.scala: ########## @@ -153,4 +155,42 @@ 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 prints its row values") { + val factory = InternalRowComparableWrapper + .getInternalRowComparableWrapperFactory(Seq(IntegerType, StringType)) + assert(factory(InternalRow(1, UTF8String.fromString("a"))).toString === "[1,a]") + assert(factory(InternalRow(null, UTF8String.fromString("a"))).toString === "[null,a]") + + // UnsafeRow's own toString is a hex dump of its words, so the wrapper has to read the values. + val unsafeProjection = UnsafeProjection.create(Array[DataType](IntegerType, StringType)) + val unsafeRow = unsafeProjection(InternalRow(2, UTF8String.fromString("b"))) + assert(factory(unsafeRow).toString === "[2,b]") + } + + test("SPARK-60859: a wrapper prints key types readably") { + // Date, Timestamp in UTC, and a binary key. + val common = InternalRowComparableWrapper.getInternalRowComparableWrapperFactory( + Seq(DateType, TimestampType, BinaryType)) + val commonRow = InternalRow(19724, 1704164645000000L, Array[Byte](1, 2)) + withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") { + assert(common(commonRow).toString === "[2024-01-02,2024-01-02 03:04:05,0x0102]") + } + + // Decimal, TimestampNTZ, Time in nanoseconds, Char and Varchar. + val other = InternalRowComparableWrapper.getInternalRowComparableWrapperFactory( + Seq(DecimalType(10, 2), TimestampNTZType, TimeType(), CharType(5), VarcharType(5))) + val otherRow = InternalRow( + Decimal(BigDecimal("12.34"), 10, 2), + 1704164645000000L, // 2024-01-02 03:04:05 + 11045000000000L, // 03:04:05 in nanoseconds since midnight + UTF8String.fromString("ab"), + UTF8String.fromString("cd")) + assert(other(otherRow).toString === "[12.34,2024-01-02 03:04:05,03:04:05,ab,cd]") + + // Two collated keys that compare equal still print their own case. + val collated = InternalRowComparableWrapper + .getInternalRowComparableWrapperFactory(Seq(StringType("UTF8_LCASE"))) + assert(collated(InternalRow(UTF8String.fromString("A"))).toString === "[A]") Review Comment: **Nit (P3):** The comment says two collated keys that compare equal still print their own case, but only `'A'` is wrapped and asserted. The test never builds `'a'`, never checks that the two wrappers are equal, and never checks that `'a'` prints as `[a]`. A renderer that upper-cased UTF8_LCASE values would still print `[A]` here and pass, while breaking the behavior the `KeyedPartitioning` scaladoc documents. **Verification:** - **Regression:** Two UTF8_LCASE keys 'a' and 'A' compare equal and render as '[a]' and '[A]', so a renderer that normalized case fails the test. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/InternalRowComparableWrapper.scala: ########## @@ -65,6 +66,11 @@ class InternalRowComparableWrapper private ( _ordering } + override def toString: String = + row.toSeq(dataTypes).zip(dataTypes) + .map { case (value, dataType) => Literal.create(value, dataType).toString } Review Comment: **Non-blocking (P2):** `Literal.create` is the factory for external values. When CHAR/VARCHAR are first-class types (`spark.sql.preserveCharVarcharTypeInfo` or `spark.sql.charVarchar.standardSemantics.enabled`), its CHAR/VARCHAR branch converts the value through `CharConverter`/`VarcharConverter`, which call `charTypeWriteSideCheck`/`varcharTypeWriteSideCheck`. A shorter CHAR key then prints right-padded, and a stored value longer than the declared length throws `EXCEED_LIMIT_LENGTH` (a `SparkRuntimeException`) while the plan is being printed. In that mode a DSv2 relation keeps the column's CHAR/VARCHAR type, so an `identity(c)` key is wrapped at that type, and `KeyedShuffleSpec.createPartitioning` carries those wrappers into the shuffled side's `Exchange`. The plan description that `SQLExecution` builds before the query runs catches only `AnalysisException` and `IllegalArgumentException`, and simple or extended `explain` catches only `AnalysisException`. Under `preserveCharVarcharTypeInfo`, reads return over-length stored values as they are, so such a join ran on the base branch, where this method printed an object identity, but now fails at query start. The new CHAR/VARCHAR assertions run with the default settings, so they don't reach this branch. This extends an [existing discussion](https://github.com/apache/spark/pull/60078#pullrequestreview-5764251521). See **Shared repair plan 1** in the review body. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/physical/partitioning.scala: ########## @@ -748,17 +748,32 @@ case class KeyedPartitioning( override def dataType: DataType = IntegerType /** - * 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. + * Prints the layout as the partition keys followed by the flags and origins, with the keys + * summarized when there are more than `maxFields` of them. `dataTypes` is left out because its + * naming is erased, so printing it would put a struct field name into a plan that appears + * nowhere in the query. Every other field of the layout is printed: the plan string is the only + * place a reader sees why a marked partitioning still shuffles. Keys print as values, so two + * keys that `equals` considers equal can print differently, e.g. `'a'` and `'A'` under + * `UTF8_LCASE`. */ - override protected def stringArgs: Iterator[Any] = - Iterator(expressions, partitionKeys, isGrouped, isCollapsed) ++ unknownKeysOrigin ++ - ungroupingOrigin + override def toString: String = simpleString(SQLConf.get.maxToStringFields) + + override def simpleString(maxFields: Int): String = { + val trailing = Seq[Any](isGrouped, isCollapsed) ++ unknownKeysOrigin ++ ungroupingOrigin + // The keys take the fields that the expressions and the trailing flags leave. + val keyBudget = maxFields - expressions.size - trailing.size + val keys = if (keyBudget < 1) { Review Comment: **Nit (P3):** When `keyBudget < 1`, this branch drops a non-empty key list without any marker, and the outer `truncatedString` then counts only the expressions and the trailing fields. With `spark.sql.debug.maxToStringFields=3`, a partitioning with 12 keys prints `keyedpartitioning(x, true, false)`, exactly what an empty key list prints, so EXPLAIN and the SQL UI show a keyed `Exchange` with keys as if it had none. At a budget of 2 it prints `keyedpartitioning(x, true, ... 1 more fields)`, which counts `isCollapsed` but not the 12 keys. The config documents that dropped elements are replaced by a `... N more fields` placeholder, and on the base branch the flattened argument list always counted the keys there. Giving the trailing fields precedence is fine; the problem is only that the dropped keys go uncounted. The new `DistributionSuite` assertions pin these unmarked strings. See **Shared repair plan 2** in the review body. -- 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]
