wombatu-kun commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3953968870
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -191,6 +242,126 @@ public void applySourceWatermark() {
"watermark-column needs to be configured to use source watermark.");
}
+ @Override
+ public boolean applyAggregates(
+ List<int[]> groupingSets,
+ List<AggregateExpression> aggregateExpressions,
+ DataType producedDataType) {
+ if (!readConf().aggregatePushDownEnabled()) {
Review Comment:
readConf() builds FlinkConfParser from table.properties(), so every batch
aggregate query loads the table from the catalog just to read a flag that is
off by default. FlinkConfParser already has a Table-free constructor and this
option has no table-property tier, so expose that through FlinkReadConf to keep
the check off the catalog.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -191,6 +242,126 @@ public void applySourceWatermark() {
"watermark-column needs to be configured to use source watermark.");
}
+ @Override
+ public boolean applyAggregates(
+ List<int[]> groupingSets,
+ List<AggregateExpression> aggregateExpressions,
+ DataType producedDataType) {
+ if (!readConf().aggregatePushDownEnabled()) {
+ LOG.info("Skipping aggregate pushdown: aggregate push down is not
enabled");
+ return false;
+ }
+
+ if (!isBounded(properties)) {
+ LOG.info("Skipping aggregate pushdown: streaming reads are not
supported");
+ return false;
+ }
+
+ if (groupingSets.size() != 1 || groupingSets.get(0).length > 0) {
+ LOG.info("Skipping aggregate pushdown: GROUP BY push down is not
supported");
+ return false;
+ }
+
+ if (limit != null) {
Review Comment:
This guards only the SQL LIMIT, while the `limit` read option and
`connector.iceberg.limit` reach the scan through FlinkReadConf.limit(), so a
pushed-down COUNT(*) answers over the whole table instead of the limited row
set. Return false as well when readConf().limit() is greater than zero.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java:
##########
@@ -129,6 +137,19 @@ private CloseableIterable<RowData> newIterable(
return iter;
}
+ private Schema filterReadSchema(Schema projected, List<Expression> filters) {
Review Comment:
In the table source path the projection can only be missing a filter column
when applyFilters already proved that filter selects whole partitions, and
those files all have a TRUE residual, so the row filter can never drop a row.
Consider keeping such filters out of the row filter so the extra columns are
not read at all; follow-up, not a blocker.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
