Visorgood commented on code in PR #6110:
URL: https://github.com/apache/datafusion-comet/pull/6110#discussion_r4079853347
##########
spark/src/test/scala/org/apache/comet/exec/CometColumnarShuffleSuite.scala:
##########
@@ -814,6 +812,86 @@ abstract class CometColumnarShuffleSuite extends
CometTestBase with AdaptiveSpar
}
}
+ // A struct containing a double is orderable by the JVM, but CometSortOrder
reports Incompatible
+ // for nested floating point under strict mode. The columnar path never
sends the sort order to
+ // native code, so that verdict must not gate it (#5971).
+ test("range partitioning on a nested floating-point key uses columnar
shuffle") {
+ withSQLConf(
+ CometConf.COMET_EXEC_STRICT_FLOATING_POINT.key -> "true",
+ CometConf.getExprAllowIncompatConfigKey("SortOrder") -> "false") {
+ withParquetTable((0 until 20).map(i => (i.toDouble, i)),
"range_struct_tbl") {
+ val df = sql("SELECT struct(_1 AS a, _2 AS b) AS c FROM
range_struct_tbl")
+ .repartitionByRange(4, col("c"))
+ checkShuffleAnswer(df, 1)
+ }
+ }
+ }
+
+ // A Scala UDF with the codegen dispatcher disabled has no serde at all, so
exprToProto returns
+ // None for it. The columnar path evaluates partition keys on the JVM
through UnsafeProjection,
+ // so that verdict must not gate the exchange (#5971). Unlike the
strictFloatingPoint cases these
+ // reproduce at default config, which is the shape that bites in practice.
+ test("range partitioning on an unserializable expression uses columnar
shuffle") {
+ withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") {
+ withParquetTable((0 until 20).map(i => (i, i.toString)),
"range_udf_tbl") {
+ val bump = udf((x: Int) => x + 1)
+ val df = sql("SELECT _1, _2 FROM range_udf_tbl").repartitionByRange(4,
bump(col("_1")))
+ checkShuffleAnswer(df, 1)
+ }
+ }
+ }
+
+ test("hash partitioning on an unserializable expression uses columnar
shuffle") {
+ withSQLConf(CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "false") {
+ withParquetTable((0 until 20).map(i => (i, i.toString)), "hash_udf_tbl")
{
+ val bump = udf((x: Int) => x + 1)
+ val df = sql("SELECT _1, _2 FROM hash_udf_tbl").repartition(4,
bump(col("_1")))
+ checkShuffleAnswer(df, 1)
+ }
+ }
+ }
+
+ /**
+ * checkShuffleAnswer only compares the query answer, which is
order-insensitive and so would
+ * pass even if Comet routed rows to different partitions than Spark. Compare
+ * spark_partition_id() per row instead.
+ */
+ private def checkPartitionAssignmentMatchesSpark(df: => DataFrame, clue:
String): Unit = {
Review Comment:
Agreed, that would have passed while testing nothing. Added
`checkCometExchange(df, 1, false)` at the top of the helper, so both assignment
tests now pin the Comet run to one exchange before comparing partition ids.
--
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]
