Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950642001
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -191,6 +241,123 @@ 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
(!readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED))
{
+ LOG.info(
+ "Skipping aggregate pushdown:
table.exec.iceberg.aggregate-push-down-enabled 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) {
+ LOG.info("Skipping aggregate pushdown: a limit is present");
+ return false;
+ }
+
+ List<Expression> icebergAggregates =
convertAggregates(aggregateExpressions);
+ if (icebergAggregates == null) {
+ return false;
+ }
+
+ Table table = loadTable();
+ if (table instanceof BaseMetadataTable) {
+ LOG.info("Skipping aggregate pushdown: metadata tables are not
supported");
+ return false;
+ }
+
+ if (!filtersSelectWholePartitions(table)) {
+ LOG.info("Skipping aggregate pushdown: a filter that doesn't select
whole partitions");
+ return false;
+ }
+
+ AggregateEvaluator evaluator = planAggregateEvaluator(table,
icebergAggregates);
+ if (evaluator == null) {
+ return false;
+ }
+
+ this.pushedAggregate = evaluator;
+ this.pushedAggregateProducedDataType = producedDataType;
+ return true;
+ }
+
+ private List<Expression> convertAggregates(List<AggregateExpression>
aggregateExpressions) {
+ List<Expression> icebergAggregates =
+ Lists.newArrayListWithExpectedSize(aggregateExpressions.size());
+ for (AggregateExpression flinkAggregate : aggregateExpressions) {
+ Expression icebergAggregate = FlinkAggregates.convert(flinkAggregate);
+ if (icebergAggregate == null) {
+ LOG.info("Skipping aggregate pushdown: unsupported aggregate {}",
flinkAggregate);
+ return null;
+ }
+
+ icebergAggregates.add(icebergAggregate);
+ }
+
+ return icebergAggregates;
+ }
+
+ private AggregateEvaluator planAggregateEvaluator(
+ Table table, List<Expression> icebergAggregates) {
+ AggregateEvaluator evaluator;
+ try {
+ evaluator = AggregateEvaluator.create(table.schema(), icebergAggregates);
+ } catch (RuntimeException e) {
+ LOG.info("Skipping aggregate pushdown: failed to bind aggregate
expressions", e);
+ return null;
+ }
+
+ if (!AggregatePushDownUtil.metricsModeSupportsAggregatePushDown(
+ table, evaluator.aggregates())) {
+ return null;
+ }
+
+ TableScan scan =
+ table
+ .newScan()
Review Comment:
Thanks for pointing it out. Keep the small change in this and return false
when any of those read options is present.
--
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]
