rdblue commented on code in PR #6622:
URL: https://github.com/apache/iceberg/pull/6622#discussion_r1103897714
##########
spark/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java:
##########
@@ -158,6 +179,129 @@ public Filter[] pushedFilters() {
return pushedFilters;
}
+ @Override
+ public boolean pushAggregation(Aggregation aggregation) {
+ if (!canPushDownAggregation(aggregation)) {
+ return false;
+ }
+
+ AggregateEvaluator aggregateEvaluator;
+ try {
+ List<Expression> aggregates =
+ Arrays.stream(aggregation.aggregateExpressions())
+ .map(agg -> SparkAggregates.convert(agg))
+ .collect(Collectors.toList());
+ aggregateEvaluator = AggregateEvaluator.create(schema, aggregates);
+ } catch (UnsupportedOperationException | IllegalArgumentException e) {
+ LOG.info("Skipped aggregate pushdown: " + e);
+ return false;
+ }
+
+ if
(!metricsModeSupportsAggregatePushDown(aggregateEvaluator.aggregates())) {
+ return false;
+ }
+
+ TableScanContext context = new TableScanContext().reportWith(((BaseTable)
table).getReporter());
+ context.setColStats(true);
+ TableScan scan = new DataTableScan(table, table.schema(), context);
+ Snapshot snapshot = readSnapshot();
+ if (snapshot == null) {
+ LOG.info("Skipped aggregate pushdown: table snapshot is null");
+ return false;
+ }
+ scan = scan.useSnapshot(snapshot.snapshotId());
+ scan = configureSplitPlanning(scan);
+
+ try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles()) {
+ List<FileScanTask> tasks = ImmutableList.copyOf(fileScanTasks);
+ for (FileScanTask task : tasks) {
+ if (!task.deletes().isEmpty()) {
+ LOG.info("Skipped aggregate pushdown: detected row level deletes.");
+ return false;
+ }
+ if (!aggregateEvaluator.update(task.file())) {
+ return false;
+ }
+ }
+ } catch (IOException e) {
+ LOG.info("Skipped aggregate pushdown: " + e);
+ return false;
+ }
+
+ pushedAggregateSchema =
+ SparkSchemaUtil.convert(new
Schema(aggregateEvaluator.resultType().fields()));
+ this.pushedAggregateRows = new InternalRow[1];
+ StructLike structLike = aggregateEvaluator.result();
+ this.pushedAggregateRows[0] =
+ new
StructInternalRow(aggregateEvaluator.resultType()).setStruct(structLike);
+ return true;
+ }
+
+ private boolean canPushDownAggregation(Aggregation aggregation) {
+ if (!(table instanceof BaseTable)) {
+ return false;
+ }
+
+ if (!readConf.aggregatePushDownEnabled()) {
+ return false;
+ }
+
+ // If group by expression is the same as the partition, the statistics
information can still
+ // be used to calculate min/max/count, will enable aggregate push down in
next phase.
+ // TODO: enable aggregate push down for partition col group by expression
+ if (aggregation.groupByExpressions().length > 0) {
+ LOG.info("Skipped aggregate pushdown: group by aggregation push down is
not supported.");
+ return false;
+ }
+
+ return true;
+ }
+
+ private Snapshot readSnapshot() {
+ Snapshot snapshot;
+ if (readConf.snapshotId() != null) {
+ snapshot = table.snapshot(readConf.snapshotId());
+ } else {
+ snapshot = table.currentSnapshot();
+ }
+
+ return snapshot;
+ }
+
+ private boolean metricsModeSupportsAggregatePushDown(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("Skipped aggregate pushdown: MetricsModes.None");
+ return false;
+ } else if (mode instanceof MetricsModes.Counts) {
+ if (aggregate.op() == Expression.Operation.MAX
+ || aggregate.op() == Expression.Operation.MIN) {
+ LOG.info(
+ "Skipped aggregate pushdown: MetricsModes.Counts, but the
aggregate is "
+ + aggregate);
+ return false;
+ }
+ } else if (mode instanceof MetricsModes.Truncate) {
+ // lower_bounds and upper_bounds may be truncated, so disable push
down
+ if (aggregate.type().typeId() == Type.TypeID.STRING) {
+ if (aggregate.op() == Expression.Operation.MAX
+ || aggregate.op() == Expression.Operation.MIN) {
+ LOG.info(
+ "Skipped aggregate pushdown: MetricsModes.Truncate, but the
aggregate is MIN/MAX on a String column.");
Review Comment:
How about `"Skipped aggregate pushdown: Cannot produce min or max from
truncated values for column %s"`?
--
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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
