sunchao commented on code in PR #6726:
URL: https://github.com/apache/datafusion-comet/pull/6726#discussion_r4242428966
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala:
##########
@@ -138,13 +138,20 @@ object CometSparkToColumnarExec extends
CometSink[SparkPlan] with DataTypeSuppor
CometScanWrapper(nativeOp, CometSparkToColumnarExec(op))
}
+ /**
+ * Declines calendar intervals at any depth. Arrow holds the time part of an
interval in
+ * nanoseconds, so `IntervalMonthDayNanoWriter` overflows on one of more
than about 292 years,
+ * which Spark accepts (#5279).
+ */
override def isTypeSupported(
dt: DataType,
name: String,
fallbackReasons: ListBuffer[String]): Boolean = dt match {
- case ArrayType(StringType, _) => true
- case MapType(StringType, StringType, _) => true
- case _: ArrayType | _: MapType => false
+ case CalendarIntervalType =>
Review Comment:
[P1] [P1] Preserve Spark-compatible decimal partitioning after collection
conversion
Removing the collection rejection makes previously correct mixed-shuffle
joins silently lose rows. With `spark.comet.convert.rdd.enabled=true`, shuffle
mode `auto`, AQE and broadcast joins disabled, and 10 shuffle partitions, join
two RDD DataFrames containing matching `DECIMAL(38,0)` keys `0..99`. Keep an
`ARRAY<INT>` column on the left and a `NullType` column on the right in the
output. This head converts the left input and chooses native shuffle, while the
right retains JVM columnar shuffle. The join returns 11 rows instead of Spark’s
100. Restoring the base gate returns all 100. `MAP<INT,INT>` reproduces the
same loss.
The existing wide-decimal guard recognizes typed Dataset conversions, and
`hashesDifferentlyFromSpark` protects shuffle-input conversion. Neither
protects this RDD conversion, so the broader gate exposes #5994 to
collection-bearing inputs that previously worked. Please extend the decimal
fallback to the affected source conversions, or make native hashing
Spark-compatible, and add a mixed-shuffle regression test.
Evidence: Reproduced after a fresh full-reactor JVM compilation on Spark
4.1.3 with the exact-head CI native library. For each i in 0..99, the left row
was `(new java.math.BigDecimal(i), Seq(i, i + 1))`, alternatively `(new
java.math.BigDecimal(i), Map(i -> (i + 1)))`. The right row was `(new
java.math.BigDecimal(i), null)` with the second field declared NullType. Both
inputs used two RDD partitions. Typed-Dataset and shuffle-input conversion were
disabled. Both head cases failed Spark answer comparison with 11 versus 100
rows. Both passed with the base conversion class restored. Reproduction and
logs:
`/tmp/review6726-root-1791737670-8lvip5yf/source/Review6726CurrentJoinProbe.scala`,
`head.log`, and `base.log`. Spark’s supported-version sources hash wide
decimals via `unscaledValue().toByteArray`; Comet’s `hash_array_decimal!`
hashes fixed-width little-endian bytes.
--
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]
