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

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -2130,6 +2130,22 @@
         "Whether to use codec pool in ORC. Disable if there are bugs with codec reuse."),
     HIVE_ICEBERG_STATS_SOURCE("hive.iceberg.stats.source", "iceberg",
         "Use stats from iceberg table snapshot for query planning. This has two values metastore and iceberg"),
+    HIVE_ICEBERG_STATS_COLLECT_PART_LEVEL("hive.iceberg.stats.collect.partlevel", false,
+        "Whether column statistics of a partitioned Iceberg table are kept per partition, letting\n" +
+        "the planner estimate from the partitions a query scans. ANALYZE writes the partitions it\n" +
+        "reads, CTAS and INSERT OVERWRITE write the partitions they replace, and a major\n" +
+        "compaction of one current-spec partition refreshes it. Plain INSERT maintains no\n" +
+        "partition-level statistics: its partitions read as stale until recomputed. After changing\n" +
+        "this, statistics of the other granularity are ignored until recomputed.\n" +
+        "A scan reads a partition's statistics whole, so what it moves is the partitions it reads\n" +
+        "times the width of the table, whatever columns it asked about. That pays where a query\n" +
+        "prunes hard, which is what these are for, and stops paying where it does not: a scan\n" +
+        "reading every partition of a wide table moves more than the table-level statistics would,\n" +
+        "however few columns it wants."),
+    HIVE_ICEBERG_STATS_MAX_SNAPSHOT_LOOKBACK("hive.iceberg.stats.max.snapshot.lookback", 20,
+        "How many snapshots a read of per partition column statistics walks back through to tell\n" +
+        "which partitions the writes since have changed. Each one costs reading the manifests it\n" +
+        "wrote, so a file further back than this cannot be judged and is not served."),
     HIVE_ICEBERG_EXPIRE_SNAPSHOT_NUMTHREADS("hive.iceberg.expire.snapshot.numthreads", 4,
         "The number of threads to be used for deleting files during expire snapshot. If set to 0 or below it uses the" +
             " default DirectExecutorService"),

iceberg/iceberg-catalog/src/main/java/org/apache/iceberg/hive/MetastoreUtil.java

@@ -22,7 +22,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.common.StatsSetupConst;
@@ -48,6 +47,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.view.BaseView;
 import org.apache.iceberg.view.SQLViewRepresentation;
@@ -119,17 +119,18 @@ public static void alterTable(
     }
   }
 
-  public static List<FieldSchema> getPartitionKeys(org.apache.iceberg.Table table, int specId) {
-    Schema schema = table.specs().get(specId).schema();
-    List<FieldSchema> hiveSchema = HiveSchemaUtil.convert(schema);
-    Map<String, String> colNameToColType = hiveSchema.stream()
-        .collect(Collectors.toMap(FieldSchema::getName, FieldSchema::getType));
-    return table.specs().get(specId).fields().stream()
-        .map(partField -> new FieldSchema(
-            schema.findColumnName(partField.sourceId()),
-            colNameToColType.get(schema.findColumnName(partField.sourceId())),
-            String.format("Transform: %s", partField.transform().toString()))
-        )
+  public static List<FieldSchema> getPartitionKeys(org.apache.iceberg.Table table) {
+    Schema schema = table.spec().schema();
+
+    return table.spec().fields().stream()
+        .map(partField -> {
+          Types.NestedField col = schema.findField(partField.sourceId());
+          return new FieldSchema(
+              col.name().toLowerCase(),  // HMS lowercases column names
+              HiveSchemaUtil.convertToTypeString(col.type()),
+              "Transform: %s".formatted(partField.transform())
+          );
+        })
         .toList();
   }
 
@@ -143,7 +144,7 @@ public static Table toHiveTable(org.apache.iceberg.Table table, Configuration co
     result.setTableType(TableType.EXTERNAL_TABLE.toString());
 
     // TODO: Revert after HIVE-29633 is fixed
-    // result.setPartitionKeys(getPartitionKeys(table, table.spec().specId()));
+    // result.setPartitionKeys(getPartitionKeys(table));
     result.setPartitionKeys(Lists.newArrayList());
 
     TableMetadata metadata = ((BaseTable) table).operations().current();

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/metasummary/PuffinStatisticsSummary.java

@@ -54,7 +54,8 @@ public void getMetaSummary(Table table, MetadataTableSummary summary) {
       StatisticsFile statsFile = statsFiles.get(0);
       List<BlobMetadata> blobMetadatas = statsFile.blobMetadata();
       if (blobMetadatas != null) {
-        builder.add(PUFFIN_STATS_BLOB, blobMetadatas.stream().map(BlobMetadata::type)
+        // a file holds one blob per partition, so the kinds it holds are named once each
+        builder.add(PUFFIN_STATS_BLOB, blobMetadatas.stream().map(BlobMetadata::type).distinct()
             .collect(Collectors.joining(",")));
       }
     }

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

@@ -505,9 +505,9 @@ private void commitTable(FileIO io, ExecutorService executor, OutputTable output
           .orElse(RewritePolicy.DEFAULT.name()));
 
       if (rewritePolicy != RewritePolicy.DEFAULT) {
-        String partitionPath = jobContexts.stream()
+        String partitionName = jobContexts.stream()
             .findAny()
-            .map(x -> x.getJobConf().get(IcebergCompactionService.PARTITION_PATH))
+            .map(x -> x.getJobConf().get(IcebergCompactionService.PARTITION_NAME))
             .orElse(null);
 
         long fileSizeThreshold = jobContexts.stream()
@@ -516,7 +516,7 @@ private void commitTable(FileIO io, ExecutorService executor, OutputTable output
             .map(Long::parseLong)
             .orElse(-1L);
 
-        commitCompaction(table, snapshotId, startTime, filesForCommit, partitionPath, fileSizeThreshold);
+        commitCompaction(table, snapshotId, startTime, filesForCommit, partitionName, fileSizeThreshold);
       } else {
         commitOverwrite(table, branchName, snapshotId, startTime, filesForCommit);
       }
@@ -623,14 +623,14 @@ private void commit(Transaction txn, SnapshotUpdate<?> update) {
    * @param snapshotId    The snapshot id of the table to use for validation
    * @param startTime     The start time of the commit - used only for logging
    * @param results       The object containing the new files
-   * @param partitionPath The path of the compacted partition
+   * @param partitionName The name of the compacted partition
    */
   private void commitCompaction(Table table, Long snapshotId, long startTime, FilesForCommit results,
-      String partitionPath, long fileSizeThreshold) {
+      String partitionName, long fileSizeThreshold) {
     List<DataFile> existingDataFiles =
-        IcebergCompactionUtil.getDataFiles(table, snapshotId, partitionPath, fileSizeThreshold);
+        IcebergCompactionUtil.getDataFiles(table, snapshotId, partitionName, fileSizeThreshold);
     List<DeleteFile> existingDeleteFiles = fileSizeThreshold == -1 ?
-        IcebergCompactionUtil.getDeleteFiles(table, snapshotId, partitionPath) : Collections.emptyList();
+        IcebergCompactionUtil.getDeleteFiles(table, snapshotId, partitionName) : Collections.emptyList();
 
     Transaction txn = IcebergAcidUtil.getOrCreateTransaction(table, jobConf);
 
@@ -644,7 +644,7 @@ private void commitCompaction(Table table, Long snapshotId, long startTime, File
     }
     commit(txn, rewriteFiles);
     LOG.info("Compaction commit took {} ms for table: {} partition: {} with {} file(s)",
-        System.currentTimeMillis() - startTime, table, StringUtils.defaultString(partitionPath, "N/A"),
+        System.currentTimeMillis() - startTime, table, StringUtils.defaultString(partitionName, "N/A"),
         results.dataFiles().size());
   }