[WIP] HIVE-29834: Iceberg: Answer min/max/count aggregates from column statistics

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

@@ -2130,6 +2130,15 @@
         "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 collected per partition.\n" +
+        "Per partition statistics let the planner estimate from the partitions a query scans, but\n" +
+        "only an ANALYZE statement writes them. INSERT and CTAS maintain the table level ones.\n" +
+        "After a change, existing statistics of the other granularity are ignored until recomputed."),
+    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 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 @@
     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/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 path 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());
   }
 

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

@@ -140,8 +140,7 @@ public void initialize(Configuration conf, Properties serDeProperties,
       }
     }
 
-    this.projectedSchema =
-        projectedSchema(conf, serDeProperties.getProperty(Catalogs.NAME), tableSchema, jobConf);
+    this.projectedSchema = projectedSchema(conf, serDeProperties, tableSchema, jobConf);
 
     if (!IcebergTableUtil.isFanoutEnabled(serDeProperties::getProperty)) {
       // ClusteredWriter requires that records are ordered by partition keys.
@@ -156,9 +155,11 @@ public void initialize(Configuration conf, Properties serDeProperties,
     }
   }
 
-  private static Schema projectedSchema(Configuration conf, String tableName, Schema tableSchema,
-      Map<String, String> jobConf) {
+  private static Schema projectedSchema(Configuration conf, Properties serDeProperties,
+      Schema tableSchema, Map<String, String> jobConf) {
+    String tableName = serDeProperties.getProperty(Catalogs.NAME);
     Context.Operation operation = HiveCustomStorageHandlerUtils.getWriteOperation(conf::get, tableName);
+
     if (operation == null) {
       jobConf.put(InputFormatConfig.CASE_SENSITIVE, "false");
       String[] selectedColumns = ColumnProjectionUtils.getReadColumnNames(conf);
@@ -180,11 +181,15 @@ private static Schema projectedSchema(Configuration conf, String tableName, Sche
     }
     boolean isCOW = IcebergTableUtil.isCopyOnWriteMode(operation, conf::get);
     if (isCOW) {
-      return getSchemaWithRowLineage(IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns()), conf);
+      return getSchemaWithRowLineage(
+          IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns(), false), conf);
     }
     switch (operation) {
       case DELETE:
-        return IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns());
+        boolean isMergeTask = HiveCustomStorageHandlerUtils.isMergeTaskEnabled(
+            key -> serDeProperties.getProperty(key, conf.get(key)),
+            tableName);
+        return IcebergAcidUtil.createSerdeSchemaForDelete(tableSchema.columns(), isMergeTask);
       case UPDATE:
         return IcebergAcidUtil.createSerdeSchemaForUpdate(tableSchema.columns());
       case OTHER: