apache / apache/gluten

[CH] sum() not support with fallback operator

Open
#9,351 0 comments 0 reactions 0 assignees View on GitHub
bug triage
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.