HIVE-29834: Iceberg: Answer MIN, MAX and COUNT from the handler's column statistic

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java

@@ -503,6 +503,9 @@ private Map<String, String> getBasicStatistics(org.apache.hadoop.hive.ql.metadat
       boolean quickStats) {
     Map<String, String> stats;
 
+    if (hmsTable.getMetaTable() != null) {
+      return Map.of();
+    }
     // For write queries where rows got modified, don't fetch from cache as values could have changed.
     Table table = getTable(hmsTable);
     Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
@@ -511,8 +514,9 @@ private Map<String, String> getBasicStatistics(org.apache.hadoop.hive.ql.metadat
       stats = emptyStatsMap();
 
     } else if (!HiveMetaHook.ICEBERG.equals(getStatsSource()) && !quickStats &&
-        hmsTable.getSnapshotRef() == null) {
-      // the metastore parameters describe the table, not a branch: use the snapshot's counters
+        hmsTable.getQualifier().isEmpty()) {
+      // the metastore holds one unversioned set of parameters describing the current table, so a
+      // branch, a tag or a point in time is not answered from it - only a plain scan is
       stats = hmsTable.getParameters();
 
     } else {
@@ -723,6 +727,17 @@ public boolean canSetColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsT
     return HiveMetaHook.ICEBERG.equals(getStatsSource());
   }
 
+  @Override
+  public boolean areColumnStatsUptoDate(org.apache.hadoop.hive.ql.metadata.Table hmsTable, List<String> colNames) {
+    if (canSetColStatistics(hmsTable)) {
+      return IcebergStoredStats.colStatsAccurate(hmsTable, colNames, conf);
+    }
+    // the metastore holds them, and its single row describes the current table: a scan of a
+    // branch, a tag, a point in time or a metadata table is not described by it
+    return hmsTable.getQualifier().isEmpty() &&
+        StatsSetupConst.areColumnStatsUptoDate(hmsTable.getParameters(), colNames);
+  }
+
   @Override
   public boolean setColStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
       Iterator<ColumnStatistics> colStats) {
@@ -812,9 +827,9 @@ private AggrStats aggrColStats(org.apache.hadoop.hive.ql.metadata.Table hmsTable
       return new AggrStats(aggregated, partNames.size());
     }
 
+    Set<String> columns = Sets.newHashSet(colNames);
     Map<String, List<ColumnStatisticsObj>> statsByPart = IcebergColStatsReader.readPart(table, statsFile,
-        partition -> partitions.contains(partition) && upToDate.test(partition),
-        Sets.newHashSet(colNames), conf);
+        partition -> partitions.contains(partition) && upToDate.test(partition), columns, conf);
 
     List<ColumnStatistics> partStats = Lists.newArrayList();
     statsByPart.forEach((partition, statsObjs) -> {
@@ -841,7 +856,7 @@ public Long getRowCount(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
     if (hmsTable.getMetaTable() != null) {
       return null;
     }
-    return getStatsSource().equals(HiveMetaHook.ICEBERG) || hmsTable.getSnapshotRef() != null ?
+    return getStatsSource().equals(HiveMetaHook.ICEBERG) || !hmsTable.getQualifier().isEmpty() ?
         snapshotRowCount(hmsTable) : metastoreRowCount(hmsTable);
   }
 
@@ -879,6 +894,14 @@ public Map<String, Long> getRowCount(org.apache.hadoop.hive.ql.metadata.Table hm
       // does not select
       return Map.of();
     }
+    // an equality delete under an unpartitioned spec applies to every data file, so no partition's
+    // entry accounts for it. By spec, not the void name: a dropped field leaves a void transform
+    boolean globalDeletes = getOrCachePartitionStats(table, snapshot).values().stream()
+        .anyMatch(stats -> table.specs().get(stats.specId()).isUnpartitioned() &&
+            stats.equalityDeleteRecordCount() > 0);
+    if (globalDeletes) {
+      return Map.of();
+    }
     Map<String, Long> rowCounts = Maps.newHashMapWithExpectedSize(partNames.size());
     collectPartitionStatsFor(table, snapshot, partNames, (partName, stats) -> {
       if (stats.equalityDeleteRecordCount() == 0 && stats.positionDeleteRecordCount() == 0) {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java

@@ -213,7 +213,7 @@ static Table getTable(Configuration configuration, Properties properties) {
     return getTable(configuration, properties, false);
   }
 
-  static Snapshot getTableSnapshot(Table table, org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+  public static Snapshot getTableSnapshot(Table table, org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
     long snapshotId = -1;
 
     if (hmsTable.getAsOfTimestamp() != null) {

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/SchemaUtils.java

@@ -20,6 +20,7 @@
 package org.apache.iceberg.mr.hive;
 
 import java.util.List;
+import java.util.Locale;
 import org.apache.hadoop.hive.ql.parse.TransformSpec;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -33,6 +34,16 @@ public class SchemaUtils {
   private SchemaUtils() {
   }
 
+  /**
+   * Returns the name the schema gives the field now, or null where it no longer has it. Lower
+   * case, as Hive keeps a column name wherever it keeps one, while an Iceberg schema keeps
+   * whatever case the table was created with.
+   */
+  public static String getColumnName(Schema schema, int fieldId) {
+    String name = schema.findColumnName(fieldId);
+    return name == null ? null : name.toLowerCase(Locale.ROOT);
+  }
+
   public static UnboundTerm<Object> toTerm(TransformSpec spec) {
     if (spec == null) {
       return null;

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/compaction/IcebergTableOptimizer.java

@@ -287,6 +287,7 @@ private Stream<Snapshot> getRelevantSnapshots(org.apache.iceberg.Table icebergTa
     return StreamSupport.stream(icebergTable.snapshots().spliterator(), false)
         .filter(s -> pastSnapshotTimeMil == null || s.timestampMillis() > pastSnapshotTimeMil)
         .filter(s -> s.timestampMillis() <= currentSnapshot.timestampMillis())
-        .filter(s -> !s.operation().equals(DataOperations.REPLACE));
+        // a snapshot written without a summary names no operation
+        .filter(s -> !DataOperations.REPLACE.equals(s.operation()));
   }
 }

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/stats/IcebergColStatsCodec.java

@@ -29,6 +29,7 @@
 import java.util.function.IntPredicate;
 import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
 import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.thrift.TDeserializer;
 import org.apache.thrift.TException;
@@ -78,49 +79,60 @@ private IcebergColStatsCodec() {
    * entry rather than once for the blob because a merge carries partitions gathered separately -
    * they need not hold the same columns, nor hold them in the same order.
    */
-  static byte[] encodeBlob(List<byte[]> parts, List<Integer> fieldIds) throws IOException {
+  static ByteBuffer encodePartBlob(List<ColumnStatisticsObj> statsObjs, List<Integer> fieldIds)
+      throws IOException {
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     DataOutputStream data = new DataOutputStream(out);
     data.writeInt(BLOB_VERSION);
-    data.writeInt(parts.size());
-    for (int i = 0; i < parts.size(); i++) {
+    data.writeInt(statsObjs.size());
+
+    for (int i = 0; i < statsObjs.size(); i++) {
+      // vectors and histograms alike: what a read wants of them it settles once they are in hand
+      byte[] entry = encodeEntry(statsObjs.get(i));
       data.writeInt(fieldIds.get(i));
-      data.writeInt(parts.get(i).length);
-      data.write(parts.get(i));
+      data.writeInt(entry.length);
+      data.write(entry);
     }
     data.flush();
-    return out.toByteArray();
+    return ByteBuffer.wrap(out.toByteArray());
   }
 
   /**
    * The entries the given places name, the rest skipped rather than copied. A scan of a wide table
-   * asks about a few of its columns, and an entry it does not want costs a read nothing beyond the
+   * asks about a few of its columns, and an entry it does not need costs a read nothing beyond the
    * length it steps over.
    */
-  static List<byte[]> decodeBlob(ByteBuffer buf, IntPredicate wanted) {
+  static List<EncodedStats> decodePartBlob(ByteBuffer buf, IntPredicate needed) {
     ByteBuffer data = buf.duplicate().order(ByteOrder.BIG_ENDIAN);
-    if (data.remaining() < Integer.BYTES || data.getInt() != BLOB_VERSION) {
+    if (data.remaining() < 2 * Integer.BYTES || data.getInt() != BLOB_VERSION) {
       return List.of();
     }
     int count = data.getInt();
-    List<byte[]> parts = Lists.newArrayListWithCapacity(count);
+    Preconditions.checkArgument(count >= 0 && count <= data.remaining() / (2 * Integer.BYTES),
+        "Column statistics blob states %s entries, of which its remaining %s bytes cannot hold " +
+            "even the field and the length each states before its own bytes",
+        count, data.remaining());
+
+    List<EncodedStats> entries = Lists.newArrayListWithCapacity(count);
     for (int i = 0; i < count; i++) {
       int fieldId = data.getInt();
       int length = data.getInt();
-      if (wanted.test(fieldId)) {
+      Preconditions.checkArgument(length >= 0 && length <= data.remaining(),
+          "Entry of field %s states %s bytes, of the %s the blob has left to read or step over",
+          fieldId, length, data.remaining());
+
+      if (needed.test(fieldId)) {
         byte[] part = new byte[length];
         data.get(part);
-        parts.add(part);
+        entries.add(new EncodedStats(fieldId, part));
       } else {
         data.position(data.position() + length);
       }
     }
-    return parts;
+    return entries;
   }
 
-  /** What the blob holds, or nothing where it was written in a shape this does not know. */
-  static List<byte[]> decodeBlob(ByteBuffer buf) {
-    return decodeBlob(buf, fieldId -> true);
+  record EncodedStats(int fieldId, byte[] bytes) {
   }
 
   /**