[spark] Support push down aggregate #4259
Conversation
|
cc @JingsongLi thank you |
| } | ||
| } | ||
|
|
||
| def supportAggregation(aggregation: Aggregation): Boolean = { |
There was a problem hiding this comment.
There is an initialize in this supportAggregation. It looks like not a support. Can you rename this to pushAggregation?
There was a problem hiding this comment.
+1, changed to pushAggregation
| @@ -38,12 +38,12 @@ abstract class PaimonBaseScanBuilder(table: Table) | |||
|
|
|||
| protected var pushed: Array[(Filter, Predicate)] = Array.empty | |||
There was a problem hiding this comment.
Can we rename this to pushedFilters?
There was a problem hiding this comment.
pushedFilters conflicts with Spark method name, rename it to pushedPredicates
| } | ||
|
|
||
| // Only support with push down partition filter | ||
| if (pushed.length != partitionFilter.length) { |
There was a problem hiding this comment.
Can we also save postScanFilters in pushFilters, and just check is empty here.
There was a problem hiding this comment.
+1, added postScanFilters
| } | ||
|
|
||
| val readBuilder = table.newReadBuilder | ||
| if (pushed.nonEmpty) { |
There was a problem hiding this comment.
just use partitionFilter?
There was a problem hiding this comment.
partitionFilters is the Spark filter, but we need Paimon's predicate..
There was a problem hiding this comment.
@ulysses-you Thanks for your contribution!
Left some comments about naming.
|
+1 |
|
Link to #2404. |
Purpose
Make
PaimonScanBuildersupportSupportsPushDownAggregates. This pr implements a local aggregator framework to eval aggregate function. For now, onlyCountStaris supported.If the aggregation can be pushed down, build a
PaimonLocalScanto avoid execute with RDD.There are some limitations:
Tests
add test
API and Format
no
Documentation