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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
