apache / apache/gluten

[CORE] Expand project when subplan is aggregation

Open
#3,440 5 comments 0 reactions 0 assignees View on GitHub
bug triage
Dominant language
Scala
Stars
1.6k
Forks
657
Avg merge
2d 21h
Merged PRs (30d)
85

Description

### Backend

CH (ClickHouse)

### Bug description

In spark rule `org.apache.spark.sql.catalyst.optimizer.CollapseProject`, project and aggregate will be collapsed.
```
plan.transformUpWithPruning(_.containsPattern(PROJECT), ruleId) {
// ...
case p @ Project(_, agg: Aggregate)
if canCollapseExpressions(p.projectList, agg.aggregateExpressions, alwaysInline) &&
canCollapseAggregate(p, agg) =>
agg.copy(aggregateExpressions = buildCleanedProjectList(
p.projectList, agg.aggregateExpressions))
// ...
}
```

In this example, `ObjectHashAggregate` will be fallback ,because `split` expresstion currently is not support.
```
# optimize plan
Aggregate [l_orderkey#978L], [split(cast(l_orderkey#978L as string), ,, 1) AS split(a, ,, 1)#996, size(collect_set(l_shipdate#988, 0, 0), true) AS b#977]
+- Project [l_orderkey#978L, l_shipdate#988]
+- Relation default.lineitem[l_orderkey#978L,l_partkey#979L,l_suppkey#980L,l_linenumber#981L...

# gluten plan
ObjectHashAggregate(keys=[l_orderkey#978L], functions=[collect_set(l_shipdate#988, 0, 0)], output=[split(a, ,, 1)#996, b#977])
+- CHNativeColumnarToRow
+- ColumnarExchange hashpartitioning(l_orderkey#978L, 5), ENSURE_REQUIREMENTS, [plan_id=550], [id=#550], [OUTPUT] ArrayBuffer(l_orderkey:LongType, buf:BinaryType), [OUTPUT] ArrayBuffer(l_orderkey:LongType, buf:BinaryType)
+- HashAggregateTransformer(keys=[l_orderkey#978L], functions=[partial_collect_set(l_shipdate#988, 0, 0)], output=[l_orderkey#978L, buf#1000])
+- NativeFileNativeScan parquet default.lineitem[l_orderkey#978L,l_shipdate#988] Batched: true, DataFilters: [], Format: Parquet, Location ...

```
It's a bad fallback.Because partial aggregate plan not fallback, In ch backend partial agg has own struct data. The struct of data not support columnToRow.

In gluten code, final aggregate plan tranform to substrait plan will apply function `applyPostProjection`. And generate tow relnode `project` and `aggRel` to backend.

Can we expand `HashAggregate` to `Project` and `HashAggregate` before `AddTransformHintRule`. It will generate a plan like
```
# before expand
ObjectHashAggregate(keys=[l_orderkey#978L], functions=[collect_set(l_shipdate#988, 0, 0)], output=[split(a, ,, 1)#996, b#977])
+- Exchange hashpartitioning(l_orderkey#978L, 5), ENSURE_REQUIREMENTS, [plan_id=542]
+- ObjectHashAggregate(keys=[l_orderkey#978L], functions=[partial_collect_set(l_shipdate#988, 0, 0)], output=[l_orderkey#978L, buf#1000])
+- FileScan parquet default.lineitem[l_orderkey#978L,l_shipdate#988] Batched: true, DataFilters: [], Format: Parquet, Location: ...

# after expand
Project(output=[split(l_orderkey#978L, ,, 1)#996, b#977]) # new plan
ObjectHashAggregate(keys=[l_orderkey#978L], functions=[collect_set(l_shipdate#988, 0, 0)], output=[l_orderkey#978L, collect_set(l_shipdate#988, 0, 0)])
+- Exchange hashpartitioning(l_orderkey#978L, 5), ENSURE_REQUIREMENTS, [plan_id=542]
+- ObjectHashAggregate(keys=[l_orderkey#978L], functions=[partial_collect_set(l_shipdate#988, 0, 0)], output=[l_orderkey#978L, buf#1000])
+- FileScan parquet default.lineitem[l_orderkey#978L,l_shipdate#988] Batched: true, DataFilters: [], Format: Parquet, Location: ...
```

If `Project` has not support expression, we can fallback `Project` only.

### Spark version

None

### Spark configurations

_No response_

### System information

_No response_

### Relevant logs

_No response_

Contributor guide

Open the contributing guide

Research direction

Trace the Spark CollapseProject behavior through Gluten's AddTransformHintRule and applyPostProjection entry points, focusing on HashAggregate handling and the ClickHouse fallback shown in the issue. Determine how the final aggregate can expose a separate Project without moving unsupported expressions into the partial aggregate, then verify the generated plan keeps the partial aggregate native and falls back only where required.

Written by the indexing model from the issue text.

Assessment

Tech stack
clickhouse, scala
Domain
backend, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
28/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.