You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "Guosmilesmile (via GitHub)" <gi...@apache.org> on 2026/09/04 14:15:41 UTC
[PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Guosmilesmile opened a new pull request, #17946:
URL: https://github.com/apache/iceberg/pull/17946
### Summary
Adds aggregate push down to the Flink Iceberg source (`IcebergTableSource`). For batch queries that aggregate the whole table without `GROUP BY` or `LIMIT`, `COUNT(*)`, `COUNT(col)`, `MAX(col)` and `MIN(col)` can now be answered from file-level metrics alone, without reading any data files.
The feature is opt-in and disabled by default via a new Flink configuration:
```sql
SET table.exec.iceberg.aggregate-push-down-enabled = true;
```
### Supported scenarios
- Batch (bounded) reads only; streaming reads never take this path.
- Query-wide aggregates without `GROUP BY` or `LIMIT`.
- `COUNT(*)`, `COUNT(col)`, `MAX(col)`, `MIN(col)`.
### When push down is skipped (falls back to a regular scan)
- `GROUP BY` or `LIMIT` present.
- Streaming/unbounded reads.
- Any filter that is not guaranteed to select whole partitions (a file may contain rows the filter would remove).
- Row-level deletes present on planned files, or a file still needs row-level filtering (non-`TRUE` residual).
- Metrics configuration cannot produce the requested aggregate for a column (e.g. count-only mode for `MIN`/`MAX`, or truncated bounds on string/binary columns).
- Metadata tables; aggregates over types that cannot be derived from file metrics.
### Alignment with Spark
Spark already supports aggregate push down via `spark.sql.iceberg.aggregate-push-down.enabled` . This change mirrors that design:
- The metrics-mode feasibility check is extracted into a shared, engine-agnostic helper in core (`AggregatePushDownUtil.metricsModeSupportsAggregatePushDown`), now used by both Spark 4.1 and Flink to avoid duplicated logic.
- The per-file validation (row-level deletes, residual, `AggregateEvaluator.update`) follows the same pattern as Spark's `pushAggregation`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950656206
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -155,7 +153,8 @@ public boolean pushAggregation(Aggregation aggregation) {
aggregateEvaluator = AggregateEvaluator.create(expressions);
- if (!metricsModeSupportsAggregatePushDown(aggregateEvaluator.aggregates())) {
+ if (!AggregatePushDownUtil.metricsModeSupportsAggregatePushDown(
Review Comment:
Since it is a small change in spark , add to all version .
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3965391140
##########
docs/docs/flink-queries.md:
##########
@@ -77,6 +77,19 @@ SET table.exec.iceberg.use-flip27-source = false;
All other SQL settings and options documented above are applicable to the FLIP-27 source.
+### Batch aggregate push down
+
+A batch query that aggregates the whole table without `GROUP BY` or `LIMIT` can be answered from
+file-level metrics without reading any data files when the following option is enabled:
+
+```sql
+SET table.exec.iceberg.aggregate-push-down-enabled = true;
+```
+
+Only `COUNT`, `MAX` and `MIN` can be derived from file metrics. The push down is skipped, and the
+query falls back to a regular scan, when it uses `GROUP BY` or `LIMIT`, when a filter does not
Review Comment:
Thanks, fix it now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950604496
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -166,18 +199,35 @@ public void applyLimit(long newLimit) {
@Override
public Result applyFilters(List<ResolvedExpression> flinkFilters) {
List<ResolvedExpression> acceptedFilters = Lists.newArrayList();
+ List<ResolvedExpression> remainingFilters = Lists.newArrayList();
List<Expression> expressions = Lists.newArrayList();
+ Table table = null;
+
for (ResolvedExpression resolvedExpression : flinkFilters) {
Optional<Expression> icebergExpression = FlinkFilters.convert(resolvedExpression);
- if (icebergExpression.isPresent()) {
- expressions.add(icebergExpression.get());
- acceptedFilters.add(resolvedExpression);
+ if (icebergExpression.isEmpty()) {
+ remainingFilters.add(resolvedExpression);
+ continue;
+ }
+
+ Expression expression = icebergExpression.get();
+ expressions.add(expression);
+ acceptedFilters.add(resolvedExpression);
+
+ if (table == null) {
+ table = loadTable();
+ }
+
+ if (ExpressionUtil.selectsPartitions(expression, table, caseSensitive)) {
+ LOG.info("Evaluating {} entirely on the Iceberg side", expression);
+ } else {
+ remainingFilters.add(resolvedExpression);
}
}
this.filters = expressions;
- return Result.of(acceptedFilters, flinkFilters);
+ return Result.of(acceptedFilters, remainingFilters);
Review Comment:
I’m planning to keep this as is for now and see what everyone thinks about the overall direction. If the PR is considered large, we can split it up later.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3964474919
##########
docs/docs/flink-queries.md:
##########
@@ -77,6 +77,19 @@ SET table.exec.iceberg.use-flip27-source = false;
All other SQL settings and options documented above are applicable to the FLIP-27 source.
+### Batch aggregate push down
+
+A batch query that aggregates the whole table without `GROUP BY` or `LIMIT` can be answered from
+file-level metrics without reading any data files when the following option is enabled:
+
+```sql
+SET table.exec.iceberg.aggregate-push-down-enabled = true;
+```
+
+Only `COUNT`, `MAX` and `MIN` can be derived from file metrics. The push down is skipped, and the
+query falls back to a regular scan, when it uses `GROUP BY` or `LIMIT`, when a filter does not
Review Comment:
The skip list omits the time travel and ref read options, metadata tables, and the metrics-mode limits - in particular MIN/MAX on a string or binary column is never pushed down, which maxOnStringIsNotPushedDown already asserts. Add those to the list so the documented behaviour matches applyAggregates.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkAggregates.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Converts a Flink {@link AggregateExpression} to an Iceberg {@link Expression} that {@link
+ * org.apache.iceberg.expressions.AggregateEvaluator} can evaluate from file-level metrics alone,
+ * without reading any data files.
+ *
+ * <p>Only {@code COUNT(*)}, {@code COUNT(col)}, {@code MAX(col)} and {@code MIN(col)} can be
+ * derived from file metrics; {@code SUM} and {@code AVG} are not tracked by Iceberg manifests and
+ * are never converted.
+ */
+public class FlinkAggregates {
+ private FlinkAggregates() {}
+
+ public static Expression convert(AggregateExpression aggregate) {
+ if (aggregate.isDistinct()
+ || aggregate.isApproximate()
+ || aggregate.getFilterExpression().isPresent()) {
+ return null;
+ }
+
+ FunctionDefinition function = aggregate.getFunctionDefinition();
+ List<FieldReferenceExpression> args = aggregate.getArgs();
+
+ // The planner hands over the instantiated aggregate function implementation (e.g.
+ // MaxAggFunction.IntMaxAggFunction). Those classes live in flink-table-planner and are hidden
+ // from connector code behind flink-table-planner-loader, so an instanceof check would throw
+ // NoClassDefFoundError in a stock distribution. Match on the class name instead, which only
+ // reflects on the runtime class already loaded by the planner and never resolves it here.
+ String functionName = function.getClass().getSimpleName();
+
+ if ("Count1AggFunction".equals(functionName)) {
+ return Expressions.countStar();
+ } else if ("CountAggFunction".equals(functionName)) {
+ return args.size() == 1 ? Expressions.count(args.get(0).getName()) : null;
+ } else if (functionName.endsWith("MaxAggFunction")) {
Review Comment:
A user-defined aggregate whose class simple name ends in MaxAggFunction or MinAggFunction matches here too, and Flink's local-aggregate rule puts no restriction on which function reaches applyAggregates, so it would be answered from file metrics as a plain MAX/MIN. Gate the match on the declaring package as well by testing getClass().getName() against the org.apache.flink.table.planner.functions.aggfunctions prefix, which reflects on the already-loaded class exactly like getSimpleName() does.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkAggregates.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.DataTypeFactory;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.functions.FunctionKind;
+import org.apache.flink.table.functions.UserDefinedFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.inference.TypeInference;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expression.Operation;
+import org.apache.iceberg.expressions.UnboundAggregate;
+import org.apache.iceberg.flink.FlinkAggregates;
+import org.junit.jupiter.api.Test;
+
+public class TestFlinkAggregates {
+
+ private static final DataType INT_TYPE = DataTypes.INT();
+
+ private static FieldReferenceExpression field(String name) {
+ return new FieldReferenceExpression(name, INT_TYPE, 0, 0);
+ }
+
+ private static AggregateExpression aggregate(
+ FunctionDefinition function, List<FieldReferenceExpression> args) {
+ return new AggregateExpression(function, args, null, INT_TYPE, false, false, false);
+ }
+
+ @Test
+ public void countStar() {
+ Expression converted = FlinkAggregates.convert(aggregate(new Count1AggFunction(), List.of()));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ assertThat(converted.op()).isEqualTo(Operation.COUNT_STAR);
+ }
+
+ @Test
+ public void countColumn() {
+ Expression converted =
+ FlinkAggregates.convert(aggregate(new CountAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.COUNT);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void max() {
+ Expression converted =
+ FlinkAggregates.convert(aggregate(new IntMaxAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.MAX);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void min() {
+ Expression converted =
+ FlinkAggregates.convert(aggregate(new IntMinAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.MIN);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void countDistinctIsNotPushedDown() {
+ AggregateExpression distinctCount =
+ new AggregateExpression(
+ new CountAggFunction(), List.of(field("id")), null, INT_TYPE, true, false, false);
+ assertThat(FlinkAggregates.convert(distinctCount)).isNull();
+ }
+
+ @Test
+ public void approximateAggregateIsNotPushedDown() {
+ AggregateExpression approxCount =
+ new AggregateExpression(
+ new CountAggFunction(), List.of(field("id")), null, INT_TYPE, false, true, false);
+ assertThat(FlinkAggregates.convert(approxCount)).isNull();
+ }
+
+ @Test
+ public void unsupportedFunctionIsNotPushedDown() {
+ Expression converted =
+ FlinkAggregates.convert(aggregate(new IntSumAggFunction(), List.of(field("amount"))));
+ assertThat(converted).isNull();
+ }
+
+ @Test
+ public void countColumnWithoutArgIsNotPushedDown() {
+ assertThat(FlinkAggregates.convert(aggregate(new CountAggFunction(), List.of()))).isNull();
+ }
+
+ // These fixtures mirror the simple class names of Flink's instantiated aggregate function
Review Comment:
flink-table-planner is on this module's test classpath (TestFlinkTableSinkExtended imports PlannerBase from it), so the real classes are not hidden here and the comment's rationale does not hold for the test. Instantiate the real Count1AggFunction, CountAggFunction and MaxAggFunction.IntMaxAggFunction instead, so the test pins the class names the converter now matches on.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkReadConf.java:
##########
@@ -36,6 +36,10 @@ public FlinkReadConf(
this.confParser = new FlinkConfParser(table, readOptions, readableConfig);
}
+ public FlinkReadConf(Map<String, String> readOptions, ReadableConfig readableConfig) {
Review Comment:
Without a Table, FlinkConfParser leaves tableProperties empty, so splitSize, splitLookback and splitFileOpenCost silently fall back to the global defaults instead of the table's read.split.* properties on an instance built this way. Add javadoc on this public constructor naming those three getters as unsupported without a Table.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3953986577
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSourceAggregatePushDown.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestBase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkTableSourceAggregatePushDown extends TestBase {
+
+ @Parameters(name = "useFlip27Source = {0}")
+ private static Object[][] parameters() {
+ return new Object[][] {
+ {false}, {true},
+ };
+ }
+
+ @Parameter(index = 0)
+ private boolean useFlip27Source;
+
+ private static final String CATALOG_NAME = "test_catalog";
+ private static final String DATABASE_NAME = "test_db";
+ private static final String TABLE_NAME = "test_table";
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ super.getTableEnv().getConfig().getConfiguration().set(CoreOptions.DEFAULT_PARALLELISM, 1);
+ super.getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE, useFlip27Source);
+ return super.getTableEnv();
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .removeConfig(FlinkConfigOptions.TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED);
+ File warehouseFile = File.createTempFile("junit", null, temporaryDirectory.toFile());
+ assertThat(warehouseFile.delete()).isTrue();
+ String warehouse = String.format("file:%s", warehouseFile);
+
+ sql(
+ "CREATE CATALOG %s WITH ('type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')",
+ CATALOG_NAME, warehouse);
+ sql("USE CATALOG %s", CATALOG_NAME);
+ sql("CREATE DATABASE %s", DATABASE_NAME);
+ sql("USE %s", DATABASE_NAME);
+ sql(
+ "CREATE TABLE %s (id INT, data VARCHAR, d DOUBLE) WITH ('write.format.default'='%s')",
+ TABLE_NAME, FileFormat.PARQUET.name());
+ sql(
+ "INSERT INTO %s VALUES (1,'iceberg',10),(2,'b',20),(3,CAST(NULL AS VARCHAR),30)",
+ TABLE_NAME);
+ }
+
+ @AfterEach
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", DATABASE_NAME, TABLE_NAME);
+ dropDatabase(DATABASE_NAME, true);
+ dropCatalog(CATALOG_NAME, true);
+ }
+
+ @TestTemplate
+ public void countStarPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(*) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(3L));
+ }
+
+ @TestTemplate
+ public void countColumnPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(data) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(2L));
+ }
+
+ @TestTemplate
+ public void maxMinPushDown() {
Review Comment:
Row-level deletes and a none metrics mode are still uncovered, and the guards added since - hasTimeTravelOrRefOptions and the aggregate-push-down-enabled read option - have no case either. Add TestTemplate cases for them, using the /*+ OPTIONS(...) */ hint form TestFlinkSourceConfig already uses from this base class.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950662502
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkAggregates.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MaxAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MinAggFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expression.Operation;
+import org.apache.iceberg.expressions.UnboundAggregate;
+import org.junit.jupiter.api.Test;
+
+public class TestFlinkAggregates {
+
+ private static final DataType INT_TYPE = DataTypes.INT();
+
+ private static FieldReferenceExpression field(String name) {
+ return new FieldReferenceExpression(name, INT_TYPE, 0, 0);
+ }
+
+ private static AggregateExpression aggregate(
+ FunctionDefinition function, List<FieldReferenceExpression> args) {
+ return new AggregateExpression(function, args, null, INT_TYPE, false, false, false);
+ }
+
+ @Test
+ public void countStar() {
+ Expression converted = FlinkAggregates.convert(aggregate(new Count1AggFunction(), List.of()));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ assertThat(converted.op()).isEqualTo(Operation.COUNT_STAR);
+ }
+
+ @Test
+ public void countColumn() {
+ Expression converted =
+ FlinkAggregates.convert(aggregate(new CountAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.COUNT);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void max() {
+ Expression converted =
+ FlinkAggregates.convert(
+ aggregate(new MaxAggFunction.IntMaxAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.MAX);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void min() {
+ Expression converted =
+ FlinkAggregates.convert(
+ aggregate(new MinAggFunction.IntMinAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.MIN);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void countDistinctIsNotPushedDown() {
+ AggregateExpression distinctCount =
+ new AggregateExpression(
+ new CountAggFunction(), List.of(field("id")), null, INT_TYPE, true, false, false);
+ assertThat(FlinkAggregates.convert(distinctCount)).isNull();
+ }
+
+ @Test
+ public void approximateAggregateIsNotPushedDown() {
+ AggregateExpression approxCount =
+ new AggregateExpression(
+ new CountAggFunction(), List.of(field("id")), null, INT_TYPE, false, true, false);
+ assertThat(FlinkAggregates.convert(approxCount)).isNull();
+ }
+
+ @Test
+ public void unsupportedFunctionIsNotPushedDown() {
Review Comment:
You are right, fix it now.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950645552
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -217,7 +399,7 @@ public DataStream<RowData> produceDataStream(
@Override
public boolean isBounded() {
- return FlinkSource.isBounded(properties);
+ return IcebergTableSource.isBounded(properties);
Review Comment:
Ok, remove the deprecated part in FlinkSource
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3954790889
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkAggregates.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink;
+
+import java.util.List;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MaxAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MinAggFunction;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Converts a Flink {@link AggregateExpression} to an Iceberg {@link Expression} that {@link
+ * org.apache.iceberg.expressions.AggregateEvaluator} can evaluate from file-level metrics alone,
+ * without reading any data files.
+ *
+ * <p>Only {@code COUNT(*)}, {@code COUNT(col)}, {@code MAX(col)} and {@code MIN(col)} can be
+ * derived from file metrics; {@code SUM} and {@code AVG} are not tracked by Iceberg manifests and
+ * are never converted.
+ */
+public class FlinkAggregates {
+ private FlinkAggregates() {}
+
+ public static Expression convert(AggregateExpression aggregate) {
+ if (aggregate.isDistinct()
+ || aggregate.isApproximate()
+ || aggregate.getFilterExpression().isPresent()) {
+ return null;
+ }
+
+ FunctionDefinition function = aggregate.getFunctionDefinition();
+ List<FieldReferenceExpression> args = aggregate.getArgs();
+
+ if (function instanceof Count1AggFunction) {
Review Comment:
This dispatches on org.apache.flink.table.planner.functions.aggfunctions types, which a stock distribution keeps inside flink-table-planner-loader and does not expose to connector code, so enabling the flag there fails with NoClassDefFoundError instead of falling back to a normal scan. Is there a way to recognise these aggregates without planner internals, or should the docs state that flink-table-planner_2.12 has to replace the loader jar?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java:
##########
@@ -48,6 +48,8 @@ private FlinkReadOptions() {}
public static final ConfigOption<Boolean> CASE_SENSITIVE_OPTION =
ConfigOptions.key(PREFIX + CASE_SENSITIVE).booleanType().defaultValue(false);
+ public static final String AGGREGATE_PUSH_DOWN_ENABLED = "aggregate-push-down-enabled";
Review Comment:
This is the only read option here declared without a connector.iceberg.<name> ConfigOption, so the Flink config spelling that works for every neighbouring read option does nothing for this one. Was the table.exec.iceberg namespace a deliberate choice here?
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -82,6 +114,8 @@ private IcebergTableSource(IcebergTableSource toCopy) {
this.limit = toCopy.limit;
this.filters = toCopy.filters;
this.readableConfig = toCopy.readableConfig;
+ this.pushedAggregate = toCopy.pushedAggregate;
Review Comment:
Flink applies each push-down ability to a fresh copy(), and this constructor carries neither cachedTable nor readConf, so a filtered aggregate query still does two catalog loads. Copy both transient fields here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950609377
##########
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();
Review Comment:
I introduced a separate `cachedTable` field.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950592335
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -109,6 +137,11 @@ private IcebergTableSource(
this.limit = limit;
this.filters = filters;
this.readableConfig = readableConfig;
+ this.caseSensitive =
+ PropertyUtil.propertyAsBoolean(
Review Comment:
Good catch, use FlinkReadConf instead of.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3954426673
##########
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:
Good catch. fix it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3948588477
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -109,6 +137,11 @@ private IcebergTableSource(
this.limit = limit;
this.filters = filters;
this.readableConfig = readableConfig;
+ this.caseSensitive =
+ PropertyUtil.propertyAsBoolean(
Review Comment:
caseSensitive is read only from the table options, so connector.iceberg.case-sensitive set through the Flink config is ignored here while ScanContext still honours it, and applyFilters then decides partition alignment under different binding rules than the scan. Resolve it with FlinkReadConf.caseSensitive(), or at least fall back to readableConfig.get(CASE_SENSITIVE_OPTION).
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -166,18 +199,35 @@ public void applyLimit(long newLimit) {
@Override
public Result applyFilters(List<ResolvedExpression> flinkFilters) {
List<ResolvedExpression> acceptedFilters = Lists.newArrayList();
+ List<ResolvedExpression> remainingFilters = Lists.newArrayList();
List<Expression> expressions = Lists.newArrayList();
+ Table table = null;
+
for (ResolvedExpression resolvedExpression : flinkFilters) {
Optional<Expression> icebergExpression = FlinkFilters.convert(resolvedExpression);
- if (icebergExpression.isPresent()) {
- expressions.add(icebergExpression.get());
- acceptedFilters.add(resolvedExpression);
+ if (icebergExpression.isEmpty()) {
+ remainingFilters.add(resolvedExpression);
+ continue;
+ }
+
+ Expression expression = icebergExpression.get();
+ expressions.add(expression);
+ acceptedFilters.add(resolvedExpression);
+
+ if (table == null) {
+ table = loadTable();
+ }
+
+ if (ExpressionUtil.selectsPartitions(expression, table, caseSensitive)) {
+ LOG.info("Evaluating {} entirely on the Iceberg side", expression);
+ } else {
+ remainingFilters.add(resolvedExpression);
}
}
this.filters = expressions;
- return Result.of(acceptedFilters, flinkFilters);
+ return Result.of(acceptedFilters, remainingFilters);
Review Comment:
applyFilters now drops partition-aligned filters from the remaining set for every query, and RowDataFileScanTaskReader grew a wider read schema to compensate - both are always on regardless of the new flag, and neither is in the summary. Split the filter push-down change into its own PR so it can be reviewed and released independently of the opt-in aggregate feature.
##########
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:
planAggregateEvaluator builds a plain table.newScan(), so with snapshot-id, as-of-timestamp, branch, tag or start-snapshot-id set the pushed-down aggregate silently answers from the current snapshot instead of the requested one. Resolve the scan the way FlinkSplitPlanner does (useSnapshot / useRef / asOfTime plus the incremental append scan), or return false when any of those read options is present.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -237,4 +419,48 @@ public DynamicTableSource copy() {
public String asSummaryString() {
return "Iceberg table source";
}
+
+ private Table loadTable() {
+ try (TableLoader tableLoader = loader.clone()) {
+ tableLoader.open();
+ return tableLoader.loadTable();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private boolean filtersSelectWholePartitions(Table table) {
+ if (filters == null || filters.isEmpty()) {
+ return true;
+ }
+
+ for (Expression filter : filters) {
+ if (!ExpressionUtil.selectsPartitions(filter, table, caseSensitive)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private Expression filterExpression() {
+ if (filters == null) {
+ return Expressions.alwaysTrue();
+ }
+
+ return filters.stream().reduce(Expressions.alwaysTrue(), Expressions::and);
+ }
+
+ private DataStream<RowData> createAggregateDataStream(StreamExecutionEnvironment execEnv) {
+ RowData row =
+ new StructRowData(pushedAggregate.resultType()).setStruct(pushedAggregate.result());
Review Comment:
MIN and MAX on a TIME column are pushed down, but the bound is Iceberg's long micros while Flink's TIME accessor is getInt, so the query dies in StructRowData with "Unknown type for int field". Build the row as a GenericRowData through RowDataUtil.convertConstant, or skip the push down for TIME columns.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -217,7 +399,7 @@ public DataStream<RowData> produceDataStream(
@Override
public boolean isBounded() {
- return FlinkSource.isBounded(properties);
+ return IcebergTableSource.isBounded(properties);
Review Comment:
This is a byte-for-byte copy of FlinkSource.isBounded and the call this change rewrites was its only remaining caller, so v2.3 now carries two copies of the same rule. If the point is to stop depending on the deprecated FlinkSource, delete the original here and make the new one private.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java:
##########
@@ -110,4 +110,12 @@ private FlinkConfigOptions() {}
SplitAssignerType.SIMPLE
+ ": simple assigner that doesn't provide any guarantee on order or locality."))
.build());
+
+ public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED =
Review Comment:
The toggle exists only as a Flink config, so unlike Spark's aggregate-push-down-enabled it cannot be set per query through a SQL hint. Add a FlinkReadOptions entry and a FlinkReadConf accessor so it follows the documented read option > Flink configuration > table property precedence.
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -155,7 +153,8 @@ public boolean pushAggregation(Aggregation aggregation) {
aggregateEvaluator = AggregateEvaluator.create(expressions);
- if (!metricsModeSupportsAggregatePushDown(aggregateEvaluator.aggregates())) {
+ if (!AggregatePushDownUtil.metricsModeSupportsAggregatePushDown(
Review Comment:
metricsModeSupportsAggregatePushDown still exists verbatim in spark/v3.5, v4.0 and v4.2, and v4.1 is no longer the default Spark version, so this edit neither removes the duplication nor lands on the tree that gets built by default. Move the Spark side to v4.2, or drop it from this PR and do the Spark de-duplication separately.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSourceAggregatePushDown.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestBase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkTableSourceAggregatePushDown extends TestBase {
+
+ @Parameters(name = "useFlip27Source = {0}")
+ private static Object[][] parameters() {
+ return new Object[][] {
+ {false}, {true},
+ };
+ }
+
+ @Parameter(index = 0)
+ private boolean useFlip27Source;
+
+ private static final String CATALOG_NAME = "test_catalog";
+ private static final String DATABASE_NAME = "test_db";
+ private static final String TABLE_NAME = "test_table";
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ super.getTableEnv().getConfig().getConfiguration().set(CoreOptions.DEFAULT_PARALLELISM, 1);
+ super.getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE, useFlip27Source);
+ return super.getTableEnv();
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .removeConfig(FlinkConfigOptions.TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED);
+ File warehouseFile = File.createTempFile("junit", null, temporaryDirectory.toFile());
+ assertThat(warehouseFile.delete()).isTrue();
+ String warehouse = String.format("file:%s", warehouseFile);
+
+ sql(
+ "CREATE CATALOG %s WITH ('type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')",
+ CATALOG_NAME, warehouse);
+ sql("USE CATALOG %s", CATALOG_NAME);
+ sql("CREATE DATABASE %s", DATABASE_NAME);
+ sql("USE %s", DATABASE_NAME);
+ sql(
+ "CREATE TABLE %s (id INT, data VARCHAR, d DOUBLE) WITH ('write.format.default'='%s')",
+ TABLE_NAME, FileFormat.PARQUET.name());
+ sql(
+ "INSERT INTO %s VALUES (1,'iceberg',10),(2,'b',20),(3,CAST(NULL AS VARCHAR),30)",
+ TABLE_NAME);
+ }
+
+ @AfterEach
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", DATABASE_NAME, TABLE_NAME);
+ dropDatabase(DATABASE_NAME, true);
+ dropCatalog(CATALOG_NAME, true);
+ }
+
+ @TestTemplate
+ public void countStarPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(*) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(3L));
+ }
+
+ @TestTemplate
+ public void countColumnPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(data) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(2L));
+ }
+
+ @TestTemplate
+ public void maxMinPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT MAX(id), MIN(id) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(3, 1));
+ }
+
+ @TestTemplate
+ public void aggregatePushDownAcrossMultipleDataFiles() {
+ enableAggregatePushDown();
+ sql("INSERT INTO %s VALUES (4,'d',40)", TABLE_NAME);
+ sql("INSERT INTO %s VALUES (5,'e',50),(6,'f',60)", TABLE_NAME);
+
+ String query = String.format("SELECT COUNT(*), MAX(id), MIN(id) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan across multiple data files")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(6L, 6, 1));
+ }
+
+ @TestTemplate
+ public void aggregatePushDownDisabledByDefault() {
+ String query = String.format("SELECT COUNT(*) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should not be pushed into the scan when disabled")
+ .doesNotContain("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(3L));
+ }
+
+ @TestTemplate
+ public void aggregatePushDownSkippedWithFilter() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(*) FROM %s WHERE id > 1", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should not be pushed into the scan when a filter is present")
+ .doesNotContain("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(2L));
+ }
+
+ @TestTemplate
+ public void aggregatePushDownWithPartitionAlignedFilter() {
+ enableAggregatePushDown();
+
+ String partitionedTable = "partitioned_table";
+ sql(
+ "CREATE TABLE %s (id INT, data VARCHAR, d DOUBLE) PARTITIONED BY (data) "
+ + "WITH ('write.format.default'='%s')",
+ partitionedTable, FileFormat.PARQUET.name());
+ try {
+ sql("INSERT INTO %s VALUES (1,'a',10),(2,'a',20),(3,'b',30),(4,'b',40)", partitionedTable);
+
+ String query = String.format("SELECT COUNT(*) FROM %s WHERE data = 'a'", partitionedTable);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan for a partition-aligned filter")
+ .contains("aggregates=[");
+ assertThat(sql(query)).hasSize(1).containsExactly(Row.of(2L));
+
+ String nonAlignedQuery =
+ String.format("SELECT COUNT(*) FROM %s WHERE id > 2", partitionedTable);
+ assertThat(explain(nonAlignedQuery))
+ .as("Local aggregate should not be pushed into the scan for a non-aligned filter")
+ .doesNotContain("aggregates=[");
+ assertThat(sql(nonAlignedQuery)).hasSize(1).containsExactly(Row.of(2L));
+ } finally {
+ sql("DROP TABLE IF EXISTS %s.%s", DATABASE_NAME, partitionedTable);
+ }
+ }
+
+ @TestTemplate
+ public void filterPushDownOnPartitionedTableWithoutAggregate() {
Review Comment:
This asserts row contents only, so it passes just as well with the old Result.of(acceptedFilters, flinkFilters) behaviour where Flink re-applies the predicate above the scan - nothing here pins the change it exists to cover. Assert the plan (no Calc carrying data = 'a' above the scan) or use the lastScanEvent hook from TableSourceTestBase.
##########
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();
Review Comment:
loadTable clones and closes a TableLoader on every call and both applyFilters and applyAggregates trigger one, so planning a filtered aggregate query now costs two catalog round-trips where applyFilters used to do no I/O at all. Cache the loaded table in a field and keep the per-filter partition-alignment result applyFilters already computed.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkAggregates.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MaxAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MinAggFunction;
+import org.apache.flink.table.types.DataType;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expression.Operation;
+import org.apache.iceberg.expressions.UnboundAggregate;
+import org.junit.jupiter.api.Test;
+
+public class TestFlinkAggregates {
+
+ private static final DataType INT_TYPE = DataTypes.INT();
+
+ private static FieldReferenceExpression field(String name) {
+ return new FieldReferenceExpression(name, INT_TYPE, 0, 0);
+ }
+
+ private static AggregateExpression aggregate(
+ FunctionDefinition function, List<FieldReferenceExpression> args) {
+ return new AggregateExpression(function, args, null, INT_TYPE, false, false, false);
+ }
+
+ @Test
+ public void countStar() {
+ Expression converted = FlinkAggregates.convert(aggregate(new Count1AggFunction(), List.of()));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ assertThat(converted.op()).isEqualTo(Operation.COUNT_STAR);
+ }
+
+ @Test
+ public void countColumn() {
+ Expression converted =
+ FlinkAggregates.convert(aggregate(new CountAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.COUNT);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void max() {
+ Expression converted =
+ FlinkAggregates.convert(
+ aggregate(new MaxAggFunction.IntMaxAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.MAX);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void min() {
+ Expression converted =
+ FlinkAggregates.convert(
+ aggregate(new MinAggFunction.IntMinAggFunction(), List.of(field("id"))));
+ assertThat(converted).isInstanceOf(UnboundAggregate.class);
+ UnboundAggregate<?> aggregate = (UnboundAggregate<?>) converted;
+ assertThat(aggregate.op()).isEqualTo(Operation.MIN);
+ assertThat(aggregate.ref().name()).isEqualTo("id");
+ }
+
+ @Test
+ public void countDistinctIsNotPushedDown() {
+ AggregateExpression distinctCount =
+ new AggregateExpression(
+ new CountAggFunction(), List.of(field("id")), null, INT_TYPE, true, false, false);
+ assertThat(FlinkAggregates.convert(distinctCount)).isNull();
+ }
+
+ @Test
+ public void approximateAggregateIsNotPushedDown() {
+ AggregateExpression approxCount =
+ new AggregateExpression(
+ new CountAggFunction(), List.of(field("id")), null, INT_TYPE, false, true, false);
+ assertThat(FlinkAggregates.convert(approxCount)).isNull();
+ }
+
+ @Test
+ public void unsupportedFunctionIsNotPushedDown() {
Review Comment:
The planner hands applyAggregates the aggfunctions implementation, so a real SUM arrives as SumAggFunction and never as BuiltInFunctionDefinitions.SUM - this asserts on a shape production never produces. Use new SumAggFunction.IntSumAggFunction() so the test covers the input the rule actually supplies.
##########
core/src/main/java/org/apache/iceberg/util/AggregatePushDownUtil.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.util;
+
+import java.util.List;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.BoundAggregate;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Helper for deciding whether aggregates can be answered from file-level metrics for aggregate push
+ * down.
+ */
+public class AggregatePushDownUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AggregatePushDownUtil.class);
+
+ private AggregatePushDownUtil() {}
+
+ public static boolean metricsModeSupportsAggregatePushDown(
Review Comment:
MetricsUtil already owns the metrics-mode helpers in core, and AggregatePushDownUtil.metricsModeSupportsAggregatePushDown repeats the same words in class and method. Move it to MetricsUtil as supportsAggregatePushDown(Table, List<BoundAggregate<?, ?>>).
##########
core/src/main/java/org/apache/iceberg/util/AggregatePushDownUtil.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.util;
+
+import java.util.List;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.BoundAggregate;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Helper for deciding whether aggregates can be answered from file-level metrics for aggregate push
+ * down.
+ */
+public class AggregatePushDownUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AggregatePushDownUtil.class);
+
+ private AggregatePushDownUtil() {}
+
+ public static boolean metricsModeSupportsAggregatePushDown(
+ Table table, List<BoundAggregate<?, ?>> aggregates) {
+ MetricsConfig config = MetricsConfig.forTable(table);
+ for (BoundAggregate<?, ?> aggregate : aggregates) {
+ String colName = aggregate.columnName();
+ if (!colName.equals("*")) {
+ MetricsModes.MetricsMode mode = config.columnMode(colName);
+ if (mode instanceof MetricsModes.None) {
+ LOG.info("Skipping aggregate pushdown: no metrics for column {}", colName);
+ return false;
+ } else if (mode instanceof MetricsModes.Counts) {
+ if (aggregate.op() == Expression.Operation.MAX
+ || aggregate.op() == Expression.Operation.MIN) {
+ LOG.info(
+ "Skipping aggregate pushdown: cannot produce min or max from count for column {}",
+ colName);
+ return false;
+ }
+ } else if (aggregate.type().typeId() == Type.TypeID.STRING
Review Comment:
The STRING/BINARY branch lost the comment explaining that the bounds may have been truncated under an earlier metrics config, which is the only reason it fires regardless of the current mode. Carry that comment over into the extracted helper.
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/FlinkAggregates.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import java.util.List;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MaxAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MinAggFunction;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Converts a Flink {@link AggregateExpression} to an Iceberg {@link Expression} that {@link
+ * org.apache.iceberg.expressions.AggregateEvaluator} can evaluate from file-level metrics alone,
+ * without reading any data files.
+ *
+ * <p>Only {@code COUNT(*)}, {@code COUNT(col)}, {@code MAX(col)} and {@code MIN(col)} can be
+ * derived from file metrics; {@code SUM} and {@code AVG} are not tracked by Iceberg manifests and
+ * are never converted.
+ */
+public class FlinkAggregates {
Review Comment:
FlinkFilters lives in org.apache.iceberg.flink and its Spark analogue SparkAggregates sits next to SparkV2Filters rather than under .source, so this converter is the odd one out. Move it to org.apache.iceberg.flink alongside FlinkFilters.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSourceAggregatePushDown.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestBase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkTableSourceAggregatePushDown extends TestBase {
+
+ @Parameters(name = "useFlip27Source = {0}")
+ private static Object[][] parameters() {
+ return new Object[][] {
+ {false}, {true},
+ };
+ }
+
+ @Parameter(index = 0)
+ private boolean useFlip27Source;
+
+ private static final String CATALOG_NAME = "test_catalog";
+ private static final String DATABASE_NAME = "test_db";
+ private static final String TABLE_NAME = "test_table";
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ super.getTableEnv().getConfig().getConfiguration().set(CoreOptions.DEFAULT_PARALLELISM, 1);
+ super.getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE, useFlip27Source);
+ return super.getTableEnv();
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .removeConfig(FlinkConfigOptions.TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED);
+ File warehouseFile = File.createTempFile("junit", null, temporaryDirectory.toFile());
+ assertThat(warehouseFile.delete()).isTrue();
+ String warehouse = String.format("file:%s", warehouseFile);
+
+ sql(
+ "CREATE CATALOG %s WITH ('type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')",
+ CATALOG_NAME, warehouse);
+ sql("USE CATALOG %s", CATALOG_NAME);
+ sql("CREATE DATABASE %s", DATABASE_NAME);
+ sql("USE %s", DATABASE_NAME);
+ sql(
+ "CREATE TABLE %s (id INT, data VARCHAR, d DOUBLE) WITH ('write.format.default'='%s')",
+ TABLE_NAME, FileFormat.PARQUET.name());
+ sql(
+ "INSERT INTO %s VALUES (1,'iceberg',10),(2,'b',20),(3,CAST(NULL AS VARCHAR),30)",
+ TABLE_NAME);
+ }
+
+ @AfterEach
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", DATABASE_NAME, TABLE_NAME);
+ dropDatabase(DATABASE_NAME, true);
+ dropCatalog(CATALOG_NAME, true);
+ }
+
+ @TestTemplate
+ public void countStarPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(*) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(3L));
+ }
+
+ @TestTemplate
+ public void countColumnPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(data) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(2L));
+ }
+
+ @TestTemplate
+ public void maxMinPushDown() {
Review Comment:
Nothing covers the paths that must refuse the push down - MIN/MAX on the VARCHAR column already in the fixture, a counts-only or none metrics mode, row-level deletes, or an Avro table whose metrics cannot answer the aggregate - and every positive case aggregates the INT column. Add a MAX(data) rejection case and at least one non-INT aggregate so the result-type conversion is exercised beyond INT and BIGINT.
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSourceAggregatePushDown.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestBase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkTableSourceAggregatePushDown extends TestBase {
Review Comment:
TableSourceTestBase in this package already provides the useFlip27Source parameter, the catalog/database/table constants, this exact getTableEnv override and a byte-identical before/clean pair, so about fifty lines here are a re-implementation. Extend it instead - it also brings the lastScanEvent and scanEventCount hooks these tests need to prove no data files were read.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile closed pull request #17946: Flink: Support aggregate push down for Iceberg table source in batch
URL: https://github.com/apache/iceberg/pull/17946
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3956103032
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkReadOptions.java:
##########
@@ -48,6 +48,8 @@ private FlinkReadOptions() {}
public static final ConfigOption<Boolean> CASE_SENSITIVE_OPTION =
ConfigOptions.key(PREFIX + CASE_SENSITIVE).booleanType().defaultValue(false);
+ public static final String AGGREGATE_PUSH_DOWN_ENABLED = "aggregate-push-down-enabled";
Review Comment:
Yes, because this is more of an optimization at the execution layer.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950651062
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkAggregates.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import java.util.List;
+import org.apache.flink.table.expressions.AggregateExpression;
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.functions.FunctionDefinition;
+import org.apache.flink.table.planner.functions.aggfunctions.Count1AggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.CountAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MaxAggFunction;
+import org.apache.flink.table.planner.functions.aggfunctions.MinAggFunction;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+
+/**
+ * Converts a Flink {@link AggregateExpression} to an Iceberg {@link Expression} that {@link
+ * org.apache.iceberg.expressions.AggregateEvaluator} can evaluate from file-level metrics alone,
+ * without reading any data files.
+ *
+ * <p>Only {@code COUNT(*)}, {@code COUNT(col)}, {@code MAX(col)} and {@code MIN(col)} can be
+ * derived from file metrics; {@code SUM} and {@code AVG} are not tracked by Iceberg manifests and
+ * are never converted.
+ */
+public class FlinkAggregates {
Review Comment:
Move to org.apache.iceberg.flink
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#issuecomment-6091627852
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3954466657
##########
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:
I'd suggest keeping the current logic as-is for now and handling it in a follow-up.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950668590
##########
flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSourceAggregatePushDown.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.table.api.TableEnvironment;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.flink.FlinkConfigOptions;
+import org.apache.iceberg.flink.TestBase;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+public class TestFlinkTableSourceAggregatePushDown extends TestBase {
+
+ @Parameters(name = "useFlip27Source = {0}")
+ private static Object[][] parameters() {
+ return new Object[][] {
+ {false}, {true},
+ };
+ }
+
+ @Parameter(index = 0)
+ private boolean useFlip27Source;
+
+ private static final String CATALOG_NAME = "test_catalog";
+ private static final String DATABASE_NAME = "test_db";
+ private static final String TABLE_NAME = "test_table";
+
+ @Override
+ protected TableEnvironment getTableEnv() {
+ super.getTableEnv().getConfig().getConfiguration().set(CoreOptions.DEFAULT_PARALLELISM, 1);
+ super.getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_FLIP27_SOURCE, useFlip27Source);
+ return super.getTableEnv();
+ }
+
+ @BeforeEach
+ public void before() throws IOException {
+ getTableEnv()
+ .getConfig()
+ .getConfiguration()
+ .removeConfig(FlinkConfigOptions.TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED);
+ File warehouseFile = File.createTempFile("junit", null, temporaryDirectory.toFile());
+ assertThat(warehouseFile.delete()).isTrue();
+ String warehouse = String.format("file:%s", warehouseFile);
+
+ sql(
+ "CREATE CATALOG %s WITH ('type'='iceberg', 'catalog-type'='hadoop', 'warehouse'='%s')",
+ CATALOG_NAME, warehouse);
+ sql("USE CATALOG %s", CATALOG_NAME);
+ sql("CREATE DATABASE %s", DATABASE_NAME);
+ sql("USE %s", DATABASE_NAME);
+ sql(
+ "CREATE TABLE %s (id INT, data VARCHAR, d DOUBLE) WITH ('write.format.default'='%s')",
+ TABLE_NAME, FileFormat.PARQUET.name());
+ sql(
+ "INSERT INTO %s VALUES (1,'iceberg',10),(2,'b',20),(3,CAST(NULL AS VARCHAR),30)",
+ TABLE_NAME);
+ }
+
+ @AfterEach
+ public void clean() {
+ sql("DROP TABLE IF EXISTS %s.%s", DATABASE_NAME, TABLE_NAME);
+ dropDatabase(DATABASE_NAME, true);
+ dropCatalog(CATALOG_NAME, true);
+ }
+
+ @TestTemplate
+ public void countStarPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(*) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(3L));
+ }
+
+ @TestTemplate
+ public void countColumnPushDown() {
+ enableAggregatePushDown();
+
+ String query = String.format("SELECT COUNT(data) FROM %s", TABLE_NAME);
+ assertThat(explain(query))
+ .as("Local aggregate should be pushed into the scan")
+ .contains("aggregates=[");
+
+ List<Row> result = sql(query);
+ assertThat(result).hasSize(1).containsExactly(Row.of(2L));
+ }
+
+ @TestTemplate
+ public void maxMinPushDown() {
Review Comment:
Add aggregatePushDownSkippedForAvroTable 、 aggregatePushDownSkippedWithCountsMetricsMode 、 maxOnDoublePushDown、maxOnStringIsNotPushedDown for this part.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950649466
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
@@ -237,4 +419,48 @@ public DynamicTableSource copy() {
public String asSummaryString() {
return "Iceberg table source";
}
+
+ private Table loadTable() {
+ try (TableLoader tableLoader = loader.clone()) {
+ tableLoader.open();
+ return tableLoader.loadTable();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ private boolean filtersSelectWholePartitions(Table table) {
+ if (filters == null || filters.isEmpty()) {
+ return true;
+ }
+
+ for (Expression filter : filters) {
+ if (!ExpressionUtil.selectsPartitions(filter, table, caseSensitive)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ private Expression filterExpression() {
+ if (filters == null) {
+ return Expressions.alwaysTrue();
+ }
+
+ return filters.stream().reduce(Expressions.alwaysTrue(), Expressions::and);
+ }
+
+ private DataStream<RowData> createAggregateDataStream(StreamExecutionEnvironment execEnv) {
+ RowData row =
+ new StructRowData(pushedAggregate.resultType()).setStruct(pushedAggregate.result());
Review Comment:
You are right, build the row as a GenericRowData through RowDataUtil.convertConstant.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3950653328
##########
core/src/main/java/org/apache/iceberg/util/AggregatePushDownUtil.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.util;
+
+import java.util.List;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.MetricsModes;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.expressions.BoundAggregate;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.types.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Helper for deciding whether aggregates can be answered from file-level metrics for aggregate push
+ * down.
+ */
+public class AggregatePushDownUtil {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AggregatePushDownUtil.class);
+
+ private AggregatePushDownUtil() {}
+
+ public static boolean metricsModeSupportsAggregatePushDown(
+ Table table, List<BoundAggregate<?, ?>> aggregates) {
+ MetricsConfig config = MetricsConfig.forTable(table);
+ for (BoundAggregate<?, ?> aggregate : aggregates) {
+ String colName = aggregate.columnName();
+ if (!colName.equals("*")) {
+ MetricsModes.MetricsMode mode = config.columnMode(colName);
+ if (mode instanceof MetricsModes.None) {
+ LOG.info("Skipping aggregate pushdown: no metrics for column {}", colName);
+ return false;
+ } else if (mode instanceof MetricsModes.Counts) {
+ if (aggregate.op() == Expression.Operation.MAX
+ || aggregate.op() == Expression.Operation.MIN) {
+ LOG.info(
+ "Skipping aggregate pushdown: cannot produce min or max from count for column {}",
+ colName);
+ return false;
+ }
+ } else if (aggregate.type().typeId() == Type.TypeID.STRING
Review Comment:
Restore the comment in the code.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3954421929
##########
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:
Add a construct for FlinkReadConf without table.
##########
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:
Add a construct for FlinkReadConf without table.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3954420755
##########
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:
Add a construct for FlinkReadConf without table.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Flink: Support aggregate push down for Iceberg table source in batch [iceberg]
Posted by "wombatu-kun (via GitHub)" <gi...@apache.org>.
wombatu-kun commented on code in PR #17946:
URL: https://github.com/apache/iceberg/pull/17946#discussion_r3953977099
##########
flink/v2.3/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java:
##########
@@ -110,4 +110,12 @@ private FlinkConfigOptions() {}
SplitAssignerType.SIMPLE
+ ": simple assigner that doesn't provide any guarantee on order or locality."))
.build());
+
+ public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_AGGREGATE_PUSH_DOWN_ENABLED =
Review Comment:
The Read options table in flink-configuration.md does not list aggregate-push-down-enabled, so the read option added here is undocumented. Add a row there alongside case-sensitive and streaming.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org