[CH] sum() not support with fallback operator
- Dominant language
- Scala
- Stars
- 1.6k
- Forks
- 657
- Avg merge
- 2d 14h
- Merged PRs (30d)
- 80
Description
### Backend
CH (ClickHouse)
### Bug description
```
withTempView("clicks") {
val df = Seq(
// small window: [00:00, 01:00), user1, 2
("2024-09-30 00:00:00", "user1"), ("2024-09-30 00:00:30", "user1"),
// small window: [01:00, 02:00), user2, 2
("2024-09-30 00:01:00", "user2"), ("2024-09-30 00:01:30", "user2"),
// small window: [03:00, 04:00), user1, 1
("2024-09-30 00:03:30", "user1"),
// small window: [11:00, 12:00), user1, 3
("2024-09-30 00:11:00", "user1"), ("2024-09-30 00:11:30", "user1"),
("2024-09-30 00:11:45", "user1")
).toDF("eventTime", "userId")
// session window: (01:00, 09:00), user1, 3 / (02:00, 07:00), user2, 2 /
// (12:00, 12:05), user1, 3
df.createOrReplaceTempView("clicks")
val aggregatedData = spark.sql(
"""
| SELECT
| session_window(small_window, '5 minutes') AS session,
| userId,
| sum(numClicks) AS numClicks
| FROM
| (
| SELECT
| window(eventTime, '1 minute') AS small_window,
| userId,
| count(*) AS numClicks
| FROM clicks
| GROUP BY window, userId
| ) cpu_small
| GROUP BY session_window, userId
|""".stripMargin)
checkAnswer(
aggregatedData,
Seq(Row("user1", 3), Row("user2", 2))
)
}
```
### Gluten version
_No response_
### Spark version
None
### Spark configurations
_No response_
### System information
_No response_
### Relevant logs
```bash
Caused by: org.apache.gluten.exception.GlutenException: Doesn't support type AggregateFunction(sum, Nullable(Int64)) for writeValue
0. Poco::Exception::Exception(String const&, int) @ 0x00000000163162b2
1. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x000000000d35b679
2. DB::Exception::Exception(PreformattedMessage&&, int) @ 0x0000000006d2ceec
3. DB::Exception::Exception(int, FormatStringHelperImpl::type>, String&&) @ 0x0000000006d33bcb
4. local_engine::CHColumnToSparkRow::convertCHColumnToSparkRow(DB::Block const&, std::unique_ptr>, std::default_delete>>> const&) @ 0x000000000d780086
5. Java_org_apache_gluten_vectorized_CHBlockConverterJniWrapper_convertColumnarToRow @ 0x0000000006d1ca55
at org.apache.gluten.vectorized.CHBlockConverterJniWrapper.convertColumnarToRow(Native Method)
at org.apache.spark.sql.execution.utils.CHExecUtil$.getRowIterFromSparkRowInfo(CHExecUtil.scala:160)
at org.apache.spark.sql.execution.utils.CHExecUtil$.c2r(CHExecUtil.scala:169)
at org.apache.spark.sql.execution.CHColumnarToRowRDD.$anonfun$f$4(CHColumnarToRowExec.scala:105)
at org.apache.gluten.utils.Arm$.withResource(Arm.scala:25)
at org.apache.gluten.metrics.GlutenTimeMetric$.millis(GlutenTimeMetric.scala:37)
at org.apache.spark.sql.execution.CHColumnarToRowRDD.$anonfun$f$2(CHColumnarToRowExec.scala:105)
at scala.collection.Iterator$$anon$10.nextCur(Iterator.scala:587)
at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:601)
at org.apache.spark.sql.execution.aggregate.MergingSessionsExec.$anonfun$doExecute$1(MergingSessionsExec.scala:71)
at org.apache.spark.sql.execution.aggregate.MergingSessionsExec.$anonfun$doExecute$1$adapted(MergingSessionsExec.scala:68)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2(RDD.scala:881)
at org.apache.spark.rdd.RDD.$anonfun$mapPartitionsWithIndexInternal$2$adapted(RDD.scala:881)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
```
Contributor guide
Research direction
Reproduce the SQL case in the issue, then trace conversion through CHBlockConverterJniWrapper, CHExecUtil.scala, CHColumnarToRowRDD, and MergingSessionsExec. Done means the session aggregation with sum(numClicks) completes without the AggregateFunction conversion error and returns the expected rows.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- clickhouse, scala, spark
- Domain
- backend, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 30/100