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: [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]
