Iceberg: PartitionsTable#partitions returns incomplete list in case of partition evolution and null partition values

Author: deniskuzZ

iceberg/iceberg-handler/src/test/results/positive/llap/iceberg_major_compaction_partition_evolution.q.out

@@ -447,6 +447,7 @@ POSTHOOK: type: SHOW COMPACTIONS
 CompactionId	Database	Table	Partition	Type	State	Worker host	Worker	Enqueue Time	Start Time	Duration(ms)	HadoopJobId	Error message	Initiator host	Initiator	Pool name	TxnId	Next TxnId	Commit Time	Highest WriteId
 #Masked#	default	ice_orc	company_id=100/dept_id=1	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
 #Masked#	default	ice_orc	company_id=100/dept_id=2	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
+#Masked#	default	ice_orc	company_id=null/dept_id=null	MAJOR	refused	MacBook-Pro.local	360	1742034058581	1742034062982	119	 --- 	 --- 	MacBook-Pro.local	manual	iceberg	0	0	0	 --- 
 #Masked#	default	ice_orc	 --- 	MAJOR	succeeded	#Masked#	manual	iceberg	0	0	0	 --- 
 PREHOOK: query: select `partition`, spec_id, content, record_count
 from default.ice_orc.files

iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/PartitionsTable.java

@@ -29,8 +29,8 @@
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.ParallelIterable;
+import org.apache.iceberg.util.PartitionMap;
 import org.apache.iceberg.util.PartitionUtil;
-import org.apache.iceberg.util.StructLikeMap;
 
 // TODO: remove class once upgraded to Iceberg v1.7.0
 
@@ -168,21 +168,26 @@ private static StaticDataTask.Row convertPartition(Partition partition) {
 
   private static Iterable<Partition> partitions(Table table, StaticTableScan scan) {
     Types.StructType partitionType = Partitioning.partitionType(table);
-    PartitionMap partitions = new PartitionMap(partitionType);
+    PartitionMap<Partition> partitions = PartitionMap.create(table.specs());
     try (CloseableIterable<ManifestEntry<? extends ContentFile<?>>> entries = planEntries(scan)) {
       for (ManifestEntry<? extends ContentFile<?>> entry : entries) {
         Snapshot snapshot = table.snapshot(entry.snapshotId());
         ContentFile<?> file = entry.file();
         StructLike partition =
             PartitionUtil.coercePartition(
                 partitionType, table.specs().get(file.specId()), file.partition());
-        partitions.get(partition).update(file, snapshot);
+        partitions
+            .computeIfAbsent(
+                file.specId(),
+                ((PartitionData) file.partition()).copy(),
+                () -> new Partition(partition, partitionType))
+            .update(file, snapshot);
       }
     } catch (IOException e) {
       throw new UncheckedIOException(e);
     }
 
-    return partitions.all();
+    return partitions.values();
   }
 
   @VisibleForTesting
@@ -241,29 +246,6 @@ private class PartitionsScan extends StaticTableScan {
     }
   }
 
-  static class PartitionMap {
-    private final StructLikeMap<Partition> partitions;
-    private final Types.StructType keyType;
-
-    PartitionMap(Types.StructType type) {
-      this.partitions = StructLikeMap.create(type);
-      this.keyType = type;
-    }
-
-    Partition get(StructLike key) {
-      Partition partition = partitions.get(key);
-      if (partition == null) {
-        partition = new Partition(key, keyType);
-        partitions.put(key, partition);
-      }
-      return partition;
-    }
-
-    Iterable<Partition> all() {
-      return partitions.values();
-    }
-  }
-
   static class Partition {
     private final PartitionData partitionData;
     private int specId;