WIP: Upgrade iceberg to 1.11

[kokila-19]

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

[deniskuzZ]

why do we hardcode in HMS modules and not using param?

[deniskuzZ]

in downstream i think it's defined in root pom

[deniskuzZ]

do we need both PartitionStats & PartitionStatistics ?

[kokila-19]

PartitionStats will be removed from 1.12.0 and PartitionsStatistics should be used in future.
I have modified where there is issues for now.
I was thinking to migrate fully to PartitionStatistics in a separate PR.

[deniskuzZ]

could you please ask in iceberg why appendStats was made private?

how about

Subject: [PATCH] patch
---
Index: iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
--- a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java	(revision 2d64db017244f258577cd0122bdc5c588395e779)
+++ b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java	(date 1784115083616)
@@ -79,7 +79,6 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.PartitionStatistics;
 import org.apache.iceberg.PartitionStatisticsFile;
-import org.apache.iceberg.PartitionStats;
 import org.apache.iceberg.Partitioning;
 import org.apache.iceberg.PartitionsTable;
 import org.apache.iceberg.Schema;
@@ -105,7 +104,6 @@
 import org.apache.iceberg.puffin.BlobMetadata;
 import org.apache.iceberg.puffin.Puffin;
 import org.apache.iceberg.puffin.PuffinReader;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
@@ -761,20 +759,16 @@
     }
 
     Evaluator evaluator = new Evaluator(partitionsTable.schema().asStruct(), filter);
-    PartitionStats result;
 
     try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles()) {
-      result = FluentIterable.from(fileScanTasks)
+      return FluentIterable.from(fileScanTasks)
           .transformAndConcat(task -> task.asDataTask().rows())
           .filter(evaluator::eval)
-          .transform(IcebergTableUtil::recordToPartitionStats)
-          .stream().reduce(IcebergTableUtil::mergePartitionStats).orElse(null);
+          .transform(BasicPartitionStats::from)
+          .stream().reduce(BasicPartitionStats::merge)
+          .orElse(BasicPartitionStats.EMPTY)
+          .toStatsMap();
     }
-
-    if (result == null) {
-      result = new PartitionStats(null, 0);
-    }
-    return toStatsMap(result);
   }
 
   static Map<String, String> toStatsMap(PartitionStatistics stats) {
@@ -787,66 +781,40 @@
     );
   }
 
-  private static Map<String, String> toStatsMap(PartitionStats stats) {
-    return ImmutableMap.of(
-        SnapshotSummary.TOTAL_DATA_FILES_PROP, String.valueOf(stats.dataFileCount()),
-        SnapshotSummary.TOTAL_RECORDS_PROP, String.valueOf(stats.dataRecordCount()),
-        SnapshotSummary.TOTAL_EQ_DELETES_PROP, String.valueOf(stats.equalityDeleteRecordCount()),
-        SnapshotSummary.TOTAL_POS_DELETES_PROP, String.valueOf(stats.positionDeleteRecordCount()),
-        SnapshotSummary.TOTAL_FILE_SIZE_PROP, String.valueOf(stats.totalDataFileSizeInBytes())
-    );
-  }
-
-  private static PartitionStats recordToPartitionStats(StructLike record) {
-    PartitionStats stats = new PartitionStats(record.get(PART_IDX, StructLike.class), -1);
-    for (int pos = 2; pos <= 7; pos++) {
-      stats.set(pos, record.get(pos, Object.class));
-    }
-    return stats;
-  }
-
-  /**
-   * Merges partition stats counters from {@code right} into {@code left}.
-   * Replicates package-private {@code PartitionStats.appendStats()} removed in Iceberg 1.11.
-   */
-  private static PartitionStats mergePartitionStats(PartitionStats left, PartitionStats right) {
-    Preconditions.checkArgument(left.specId() == right.specId(), "Spec IDs must match");
-    left.set(PartitionStatistics.DATA_RECORD_COUNT_POSITION,
-        left.dataRecordCount() + right.dataRecordCount());
-    left.set(PartitionStatistics.DATA_FILE_COUNT_POSITION,
-        left.dataFileCount() + right.dataFileCount());
-    left.set(PartitionStatistics.TOTAL_DATA_FILE_SIZE_IN_BYTES_POSITION,
-        left.totalDataFileSizeInBytes() + right.totalDataFileSizeInBytes());
-    left.set(PartitionStatistics.POSITION_DELETE_RECORD_COUNT_POSITION,
-        left.positionDeleteRecordCount() + right.positionDeleteRecordCount());
-    left.set(PartitionStatistics.POSITION_DELETE_FILE_COUNT_POSITION,
-        left.positionDeleteFileCount() + right.positionDeleteFileCount());
-    left.set(PartitionStatistics.EQUALITY_DELETE_RECORD_COUNT_POSITION,
-        left.equalityDeleteRecordCount() + right.equalityDeleteRecordCount());
-    left.set(PartitionStatistics.EQUALITY_DELETE_FILE_COUNT_POSITION,
-        left.equalityDeleteFileCount() + right.equalityDeleteFileCount());
-    left.set(PartitionStatistics.DV_COUNT_POSITION, left.dvCount() + right.dvCount());
-
-    Long rightTotalRecords = right.totalRecords();
-    if (rightTotalRecords != null) {
-      Long leftTotalRecords = left.totalRecords();
-      if (leftTotalRecords == null) {
-        left.set(PartitionStatistics.TOTAL_RECORD_COUNT_POSITION, rightTotalRecords);
-      } else {
-        left.set(PartitionStatistics.TOTAL_RECORD_COUNT_POSITION,
-            leftTotalRecords + rightTotalRecords);
-      }
+  /** The subset of partition stats counters Hive reports, summed across PARTITIONS metadata table rows. */
+  private record BasicPartitionStats(
+      long recordCount, long fileCount, long totalFileSizeInBytes,
+      long positionDeleteRecordCount, long equalityDeleteRecordCount) {
+
+    static final BasicPartitionStats EMPTY = new BasicPartitionStats(0, 0, 0, 0, 0);
+
+    static BasicPartitionStats from(StructLike row) {
+      return new BasicPartitionStats(
+          row.get(2, Long.class),      // record_count
+          row.get(3, Integer.class),   // file_count
+          row.get(4, Long.class),      // total_data_file_size_in_bytes
+          row.get(5, Long.class),      // position_delete_record_count
+          row.get(7, Long.class));     // equality_delete_record_count
+    }
+
+    BasicPartitionStats merge(BasicPartitionStats other) {
+      return new BasicPartitionStats(
+          recordCount + other.recordCount,
+          fileCount + other.fileCount,
+          totalFileSizeInBytes + other.totalFileSizeInBytes,
+          positionDeleteRecordCount + other.positionDeleteRecordCount,
+          equalityDeleteRecordCount + other.equalityDeleteRecordCount);
+    }
+
+    Map<String, String> toStatsMap() {
+      return ImmutableMap.of(
+          SnapshotSummary.TOTAL_DATA_FILES_PROP, String.valueOf(fileCount),
+          SnapshotSummary.TOTAL_RECORDS_PROP, String.valueOf(recordCount),
+          SnapshotSummary.TOTAL_EQ_DELETES_PROP, String.valueOf(equalityDeleteRecordCount),
+          SnapshotSummary.TOTAL_POS_DELETES_PROP, String.valueOf(positionDeleteRecordCount),
+          SnapshotSummary.TOTAL_FILE_SIZE_PROP, String.valueOf(totalFileSizeInBytes)
+      );
     }
-
-    Long rightLastUpdatedAt = right.lastUpdatedAt();
-    if (rightLastUpdatedAt != null) {
-      Long leftLastUpdatedAt = left.lastUpdatedAt();
-      if (leftLastUpdatedAt == null || rightLastUpdatedAt > leftLastUpdatedAt) {
-        left.set(PartitionStatistics.LAST_UPDATED_AT_POSITION, rightLastUpdatedAt);
-        left.set(PartitionStatistics.LAST_UPDATED_SNAPSHOT_ID_POSITION, right.lastUpdatedSnapshotId());
-      }
-    }
-    return left;
   }
 
   public static PartitionSpec getPartitionSpec(Table icebergTable, String partitionPath)

[kokila-19]

could you please ask in iceberg why appendStats was made private?
apache/iceberg#14119 -> says they are modifying deprecated method to package-private.
Also, appendStats() is in PartitionStats.java which will be removed from 1.12.0 onwards. So we will to update them as well later.

[deniskuzZ]

so no need to fetch the statsFile ?

[kokila-19]

https://github.com/apache/iceberg/blob/5f3d3c560a956b8ca3e45c525a353c77f8a1f9b4/core/src/main/java/org/apache/iceberg/BasePartitionStatisticsScan.java#L76

Yes, With the introduction of newPartitionStatisticsScan(), Iceberg handles reading partition statistics file internally in BasePartitionStatisticsScan, we no longer need statsFile for I/O.
The only reason we still call getPartitionStatsFile() is the condition check, to preserve existing behavior when no stats file exists in table metadata.

I’m thinking of replacing the statsFile variable with inline getPartitionStatsFile() != null or partitionStatisticsFiles(), so we don't have to fetch/hold statsFile.

[sonarqubecloud]

Quality Gate passed

Issues
3 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud