deniskuzZ commented on code in PR #6603:
URL: https://github.com/apache/hive/pull/6603#discussion_r3586699715
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java:
##########
@@ -796,6 +805,50 @@ private static PartitionStats
recordToPartitionStats(StructLike record) {
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) {
Review Comment:
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)
````
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
