[CORE] Expand project when subplan is aggregation
- 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
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