push down min/max/count to iceberg

api/src/main/java/org/apache/iceberg/TableScan.java

@@ -101,4 +101,13 @@ default TableScan appendsAfter(long fromSnapshotId) {
    * @return the Snapshot this scan will use
    */
   Snapshot snapshot();
+
+  /**
+   * Create a new {@link TableScan} from this scan's configuration that will have column stats
+   *
+   * @return a new scan based on this with column stats
+   */
+  default TableScan withColStats() {
+    throw new UnsupportedOperationException("scan with colStats is not supported");
+  }
 }

api/src/main/java/org/apache/iceberg/expressions/AggregateEvaluator.java

@@ -0,0 +1,125 @@
+/*
+ * 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.expressions;
+
+import java.util.List;
+import java.util.stream.Collectors;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.expressions.BoundAggregate.Aggregator;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+
+/**
+ * A class for evaluating aggregates. It evaluates each of the aggregates and updates the aggregated
+ * value. The final aggregated result can be returned by {@link #result()}.
+ */
+public class AggregateEvaluator {
+
+  public static AggregateEvaluator create(Schema schema, List<Expression> aggregates) {
+    return create(schema.asStruct(), aggregates);
+  }
+
+  public static AggregateEvaluator create(List<BoundAggregate<?, ?>> aggregates) {
+    return new AggregateEvaluator(aggregates);
+  }
+
+  private static AggregateEvaluator create(Types.StructType struct, List<Expression> aggregates) {
+    List<BoundAggregate<?, ?>> boundAggregates =
+        aggregates.stream()
+            .map(expr -> Binder.bind(struct, expr))
+            .map(bound -> (BoundAggregate<?, ?>) bound)
+            .collect(Collectors.toList());
+
+    return new AggregateEvaluator(boundAggregates);
+  }
+
+  private final List<Aggregator<?>> aggregators;
+  private final Types.StructType resultType;
+  private final List<BoundAggregate<?, ?>> aggregates;
+
+  private AggregateEvaluator(List<BoundAggregate<?, ?>> aggregates) {
+    ImmutableList.Builder<Aggregator<?>> aggregatorsBuilder = ImmutableList.builder();
+    List<Types.NestedField> resultFields = Lists.newArrayList();
+
+    for (int pos = 0; pos < aggregates.size(); pos += 1) {
+      BoundAggregate<?, ?> aggregate = aggregates.get(pos);
+      aggregatorsBuilder.add(aggregate.newAggregator());
+      resultFields.add(Types.NestedField.optional(pos, aggregate.describe(), aggregate.type()));
+    }
+
+    this.aggregators = aggregatorsBuilder.build();
+    this.resultType = Types.StructType.of(resultFields);
+    this.aggregates = aggregates;
+  }
+
+  public void update(StructLike struct) {
+    for (Aggregator<?> aggregator : aggregators) {
+      aggregator.update(struct);
+    }
+  }
+
+  public void update(DataFile file) {
+    for (Aggregator<?> aggregator : aggregators) {
+      aggregator.update(file);
+    }
+  }
+
+  public Types.StructType resultType() {
+    return resultType;
+  }
+
+  public boolean allAggregatorsValid() {
+    return aggregators.stream().allMatch(BoundAggregate.Aggregator::isValid);
+  }
+
+  public StructLike result() {
+    Object[] results =
+        aggregators.stream().map(BoundAggregate.Aggregator::result).toArray(Object[]::new);
+    return new ArrayStructLike(results);
+  }
+
+  public List<BoundAggregate<?, ?>> aggregates() {
+    return aggregates;
+  }
+
+  private static class ArrayStructLike implements StructLike {
+    private final Object[] values;
+
+    private ArrayStructLike(Object[] values) {
+      this.values = values;
+    }
+
+    public int size() {
+      return values.length;
+    }
+
+    @Override
+    public <T> T get(int pos, Class<T> javaClass) {
+      return javaClass.cast(values[pos]);
+    }
+
+    @Override
+    public <T> void set(int pos, T value) {
+      values[pos] = value;
+    }
+  }
+}

api/src/main/java/org/apache/iceberg/expressions/BoundAggregate.java

@@ -18,18 +18,37 @@
  */
 package org.apache.iceberg.expressions;
 
+import java.util.Map;
+import org.apache.iceberg.DataFile;
 import org.apache.iceberg.StructLike;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 
 public class BoundAggregate<T, C> extends Aggregate<BoundTerm<T>> implements Bound<C> {
+
   protected BoundAggregate(Operation op, BoundTerm<T> term) {
     super(op, term);
   }
 
   @Override
   public C eval(StructLike struct) {
-    throw new UnsupportedOperationException(this.getClass().getName() + " does not implement eval");
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement eval(StructLike)");
+  }
+
+  C eval(DataFile file) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement eval(DataFile)");
+  }
+
+  boolean hasValue(DataFile file) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement hasValue(DataFile)");
+  }
+
+  Aggregator<C> newAggregator() {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement newAggregator()");
   }
 
   @Override
@@ -44,4 +63,105 @@ public Type type() {
       return term().type();
     }
   }
+
+  public String columnName() {
+    if (op() == Operation.COUNT_STAR) {
+      return "*";
+    } else {
+      return ref().name();
+    }
+  }
+
+  public String describe() {
+    switch (op()) {
+      case COUNT_STAR:
+        return "count(*)";
+      case COUNT:
+        return "count(" + ExpressionUtil.describe(term()) + ")";
+      case MAX:
+        return "max(" + ExpressionUtil.describe(term()) + ")";
+      case MIN:
+        return "min(" + ExpressionUtil.describe(term()) + ")";
+      default:
+        throw new UnsupportedOperationException("Unsupported aggregate type: " + op());
+    }
+  }
+
+  <V> V safeGet(Map<Integer, V> map, int key) {
+    return safeGet(map, key, null);
+  }
+
+  <V> V safeGet(Map<Integer, V> map, int key, V defaultValue) {
+    if (map != null) {
+      return map.getOrDefault(key, defaultValue);
+    }
+
+    return null;
+  }
+
+  interface Aggregator<R> {
+    void update(StructLike struct);
+
+    void update(DataFile file);
+
+    boolean hasValue(DataFile file);
+
+    R result();
+
+    boolean isValid();
+  }
+
+  abstract static class NullSafeAggregator<T, R> implements Aggregator<R> {
+    private final BoundAggregate<T, R> aggregate;
+    private boolean isValid = true;
+
+    NullSafeAggregator(BoundAggregate<T, R> aggregate) {
+      this.aggregate = aggregate;
+    }
+
+    protected abstract void update(R value);
+
+    protected abstract R current();
+
+    @Override
+    public void update(StructLike struct) {
+      R value = aggregate.eval(struct);
+      if (value != null) {
+        update(value);
+      }
+    }
+
+    @Override
+    public boolean hasValue(DataFile file) {
+      return aggregate.hasValue(file);
+    }
+
+    @Override
+    public void update(DataFile file) {
+      if (isValid) {
+        if (hasValue(file)) {
+          R value = aggregate.eval(file);
+          if (value != null) {
+            update(value);
+          }
+        } else {
+          this.isValid = false;
+        }
+      }
+    }
+
+    @Override
+    public R result() {
+      if (!isValid) {
+        return null;
+      }
+
+      return current();
+    }
+
+    @Override
+    public boolean isValid() {
+      return this.isValid;
+    }
+  }
 }

api/src/main/java/org/apache/iceberg/expressions/CountAggregate.java

@@ -0,0 +1,71 @@
+/*
+ * 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.expressions;
+
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.StructLike;
+
+public class CountAggregate<T> extends BoundAggregate<T, Long> {
+  protected CountAggregate(Operation op, BoundTerm<T> term) {
+    super(op, term);
+  }
+
+  @Override
+  public Long eval(StructLike struct) {
+    return countFor(struct);
+  }
+
+  @Override
+  public Long eval(DataFile file) {
+    return countFor(file);
+  }
+
+  protected Long countFor(StructLike row) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement countFor(StructLike)");
+  }
+
+  protected Long countFor(DataFile file) {
+    throw new UnsupportedOperationException(
+        this.getClass().getName() + " does not implement countFor(DataFile)");
+  }
+
+  @Override
+  public Aggregator<Long> newAggregator() {
+    return new CountAggregator<>(this);
+  }
+
+  private static class CountAggregator<T> extends NullSafeAggregator<T, Long> {
+    private Long count = 0L;
+
+    CountAggregator(BoundAggregate<T, Long> aggregate) {
+      super(aggregate);
+    }
+
+    @Override
+    protected void update(Long value) {
+      count += value;
+    }
+
+    @Override
+    protected Long current() {
+      return count;
+    }
+  }
+}