Add aggregation pushdown support for count using Iceberg Metrics

[osscm]

Description

Iceberg maintains metrics of the data files in the Iceberg metadata. So, we can utilize these metrics to optimize the simple aggregate queries.
To achieve this, Query Engine needs to support aggregation pushdown at the connector layer. Trino connector framework supports aggregation pushdown but Iceberg and Hive connectors have not implemented it. Aggregation Pushdown.
If Iceberg connector can support it then we can use the Iceberg metrics to give some of the answers like count, min, and max with the partition filters. Currently, in Iceberg and Hive connectors, aggregations are handled at the engine level.
Important points
This is a single-threaded implementation and generates a single split to populate the page/data for the result.
We might be able to implement it in a distributed manner If an engine can support handling aggregation at the top level while handing the aggregation pushdown. Like how the engine handles count(*) by summing the result/count it gets from different workers.
This implementation will also not handle the situation where the count is not present for some files/partitions so that some metrics can be used partially, and for some data still, the metrics will be read from the actual data files.
My understanding is, that this would require the following feature to be done: #13534
It can handle positional delete but not equality-based deletes.

Additional context and related issues

ConnectorMetadata.applyAggregation(....) can result into projected column per aggregate function, if it needs to pushed down

applyAggregation is called by the QueryEngine in and latter on in the flow, it calls the ConnectorSplitManager to generate the splits.

IcebergSplitManager creates the IcebergSplitSource now it can create AggregateSplitSource as well.

AggregateSplitSource (like IcebergSplitSource) apply the partition filter and get the dataFiles needed, and then use the metrics. Now it can create AggregationSplit with the calculated count/metrics.

IcebergPageSourceProvider.createPageSource gets the split, now here if aggregation pushdown is used then create the AggregatePageSource based on the count/min/max available in the split.

This feature is controlled by session level flag aggregation_pushdown_enabled

Following type of queries can be benefitted

SELECT count(*) FROM tbl ;

SELECT count(*) FROM tbl WHERE part_col = 'foo' ;

SELECT count(*) FROM tbl WHERE part_date =DATE  '2023-01-10' ;
SELECT count(*) FROM tbl WHERE part_date BETWEEN DATE  '2023-01-10'  AND DATE  '2023-01-15' ;

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`15745`)

For #10974

[osscm]

To understand some of the class dependencies.

[osscm]

Will/Can refactor to avoid the duplicate code of IcebergSplitSource, as it's based on it.
But first thought of getting feedback on the approach, then will do further refactoring.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[mosabua]

@colebow should the stalebot ignore drafts maybe?

@osscm are you still working on this?

[bitsondatadev]

@colebow should the stalebot ignore drafts maybe?

I would argue it should still look at these. Many drafts are started and abandoned. I've closed out some already that weren't going to be worked on.

[mosabua]

Fair .. lets keep drafts in..

[osscm]

@mosabua @bitsondatadev yes Im working on it.
@alexjo2144 and I had discussed about the initial design.
So, put together this draft, and wanted to get his feedback.

Alex, whenever you get time, can you please check.

[bitsondatadev]

Thanks for that background @osscm, I'll flag Alex if he doesn't look at this within the next few days!

[osscm]

I'll check the conflicts.

@alexjo2144 thanks a lot for the discussions.
I have started the work to support the min/max functions, so it would be great if we can catchup on this one.

[osscm]

Hi Eric, Thanks for your interest. I’ll resume, have to rebase them PR. Also, had a good discussion with the team, and requested if someone can help to review the PR. So hoping to see progress soon. Internally we have started working on the min/max as well.

…

On Thu, Jun 15, 2023 at 2:25 PM Erin Drummond [BitMEX] < ***@***.***> wrote: Any progress on this? I'm hanging out for this feature :) — Reply to this email directly, view it on GitHub <#15832 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AXQ2PYTQQ5HANHMRMHTIW3DXLN4TNANCNFSM6AAAAAAUFZ72TY> . You are receiving this because you were mentioned.Message ID: ***@***.***>

[mosabua]

Can you fix the checkstyle errors and the look at the test failure @osscm

Also @martint @findepi @alexjo2144 .. can you have a look at this

[vgankidi]

Hi team, could you please confirm if the approach looks good here and you have time to review this. We have added support for additional aggregation pushdowns and would like to contribute that as well once this goes through. Once we have confirmation on the approach we can go ahead and rebase this again if someone can take a look at it. As we progress, constant rebasing takes time too, so would appreciate feedback on the overall approach first. cc @martint @findepi @electrum @alexjo2144 @osscm

[findepi]

for count(*) pushdown we might be able to do something simpler
for example see the code in Delta connector --

trino/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakePageSourceProvider.java

Lines 186 to 198 in 6f15559

	(regularColumns.isEmpty() || onlyRowIdColumn(regularColumns))) {
	return new DeltaLakePageSource(
	deltaLakeColumns,
	ImmutableSet.of(),
	partitionKeys,
	partitionValues,
	generatePages(split.getFileRowCount().get(), onlyRowIdColumn(regularColumns)),
	Optional.empty(),
	split.getPath(),
	split.getFileSize(),
	split.getFileModifiedTime(),
	Optional::empty);
	}

[amogh-jahagirdar]

I think we need some tests which validate that agg pushdown actually happened by looking at the explain plan?

[osscm]

I don't think I'd use anything that's not spec'd out as a source of truth for determining if equality deletes exist.
@amogh-jahagirdar thanks for the giving time for the discussion.
it was nice talking to you.

As we discussed, we will go ahead with this property check, and you will also check with the Iceberg's team to add this property to be part of the Spec, as it is already being updated.
Snapshot Summary is already part of the spec.

I have also discussed it with @flyrain (Iceberg team), and he also gave +1 for this.
thanks @flyrain

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[mosabua]

Please rebase @osscm

Can @electrum and @findepi please figure out next steps.

[github-actions]

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

[github-actions]

Closing this pull request, as it has been stale for six weeks. Feel free to re-open at any time.

[mosabua]

We really need to get this in .. cc @alexjo2144 @findepi @findinpath @cwsteinbach @dain ..

I am going to add the stale-ignore label so we wont have it closed again.

[sopel39]

Given discussion in https://trinodb.slack.com/archives/CJ6UC075E/p1722544669964209 I wonder how we can proceed here. cc @dain @findepi @raunaqmorarka
It seems Iceberg spec explicitly allows deviation from accurate https://iceberg.apache.org/spec/#manifests stats. I wonder if we can detect that the stats are in fact correct. Maybe we miss something like https://issues.apache.org/jira/browse/PARQUET-2352?

[sopel39]

@findepi @dain would it be fine to land it behind feature flag (disabled by default)? Additionally, we would like to implement similar logic as for Spark (https://github.com/apache/iceberg/blob/main/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java#L309). It's not perfect, so in the meantime we could work with Iceberg community to have more firm spec for column stats accuracy. Once this is done, we could enable it by default.

[findepi]

@sopel39 there are basic unresolved comments at this point, so there is probably no point in another review round, least to talk about landing.
Also, it's not clear we've settled on the strategy how this should work on the logical level. For example, do we want to violate iceberg spec and take upper bound as max value? Or do we want to use lower/upper bounds for pruning only. The latter could typically be similarly efficient but without violating iceberg spec, so would pave the way to eventually enable the feature without a flag.

[sopel39]

Also, it's not clear we've settled on the strategy how this should work on the logical level.

@findepi that's exactly what I mean rather than landing without another review round.

For example, do we want to violate iceberg spec and take upper bound as max value?

It seems Iceberg itself is violating it's own spec with Spark unfortunately regarding to min/max stats

Or do we want to use lower/upper bounds for pruning only. The latter could typically be similarly efficient but without violating iceberg spec

Don't we prune already using Iceberg stats? Anyways, pruning and agg pushdown are completely different things.

Also, we talked about min/max but this PR is abount count. According to spec count stats are indeed accurate:
https://iceberg.apache.org/spec/#manifest-lists

513 existing_rows_count

Number of rows in all of files in the manifest that have status EXISTING, when null this is assumed to be non-zero

So we should be good to just have it enabled for counts for the moment

[findepi]

Don't we prune already using Iceberg stats? Anyways, pruning and agg pushdown are completely different things.

Yes to both

Also, we talked about min/max but this PR is abount count. According to spec count stats are indeed accurate:

You're right. I will copy #15832 (comment) here:

Agreed. However, we need to understand what the approach would look like to support other aggregations than count, since the input information, and guarantees, is so much different.

[sopel39]

Agreed. However, we need to understand what the approach would look like to support other aggregations than count, since the input information, and guarantees, is so much different.

I think to get guarantees for other stats Iceberg spec has to be updated. My suggestion would be some kind of optional field in the metadata itself specifying if stats are estimates or not. cc @osscm

[findepi]

I think to get guarantees for other stats Iceberg spec has to be updated. My suggestion would be some kind of optional field in the metadata itself specifying if stats are estimates or not

Just like @raunaqmorarka did for Parquet footer. i like the idea.

Note that it's not must-have to do some aggregation pushdown on top of iceberg.
It's just a must-have for a particular approach we seem to be thinking about.
But we can do agg pushdown without changes to Iceberg spec.

[sopel39]

@findepi I've create Iceberg spec proposal: apache/iceberg#10930

[mosabua]

@findepi I've create Iceberg spec proposal: apache/iceberg#10930

fyi @cwsteinbach and @alexjo2144