Core: Make all of the BaseFile stat maps immutable

core/src/main/java/org/apache/iceberg/BaseFile.java

@@ -21,6 +21,7 @@
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -444,12 +445,12 @@ public Map<Integer, Long> nanValueCounts() {
 
   @Override
   public Map<Integer, ByteBuffer> lowerBounds() {
-    return toReadableMap(lowerBounds);
+    return toReadableByteBufferMap(lowerBounds);
   }
 
   @Override
   public Map<Integer, ByteBuffer> upperBounds() {
-    return toReadableMap(upperBounds);
+    return toReadableByteBufferMap(upperBounds);
   }
 
   @Override
@@ -473,10 +474,22 @@ public Integer sortOrderId() {
   }
 
   private static <K, V> Map<K, V> toReadableMap(Map<K, V> map) {
-    if (map instanceof SerializableMap) {
+    if (map == null) {
+      return null;
+    } else if (map instanceof SerializableMap) {
       return ((SerializableMap<K, V>) map).immutableMap();
     } else {
-      return map;
+      return Collections.unmodifiableMap(map);
+    }
+  }
+
+  private static Map<Integer, ByteBuffer> toReadableByteBufferMap(Map<Integer, ByteBuffer> map) {
+    if (map == null) {
+      return null;
+    } else if (map instanceof SerializableByteBufferMap) {
+      return ((SerializableByteBufferMap) map).immutableMap();
+    } else {
+      return Collections.unmodifiableMap(map);
     }
   }
 

core/src/main/java/org/apache/iceberg/SerializableByteBufferMap.java

@@ -22,13 +22,15 @@
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.ByteBuffers;
 
 class SerializableByteBufferMap implements Map<Integer, ByteBuffer>, Serializable {
   private final Map<Integer, ByteBuffer> wrapped;
+  private transient volatile Map<Integer, ByteBuffer> immutableMap;
 
   static Map<Integer, ByteBuffer> wrap(Map<Integer, ByteBuffer> map) {
     if (map == null) {
@@ -88,6 +90,18 @@ Object writeReplace() throws ObjectStreamException {
     return new MapSerializationProxy(keys, values);
   }
 
+  public Map<Integer, ByteBuffer> immutableMap() {
+    if (immutableMap == null) {
+      synchronized (this) {
+        if (immutableMap == null) {
+          immutableMap = Collections.unmodifiableMap(wrapped);
+        }
+      }
+    }
+
+    return immutableMap;
+  }
+
   @Override
   public int size() {
     return wrapped.size();

core/src/test/java/org/apache/iceberg/TestManifestReaderStats.java

@@ -27,6 +27,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.types.Conversions;
 import org.apache.iceberg.types.Types;
+import org.assertj.core.api.Assertions;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -220,6 +221,42 @@ private void assertFullStats(DataFile dataFile) {
     Assert.assertEquals(LOWER_BOUNDS, dataFile.lowerBounds());
     Assert.assertEquals(UPPER_BOUNDS, dataFile.upperBounds());
 
+    if (dataFile.valueCounts() != null) {
+      Assertions.assertThatThrownBy(
+              () -> dataFile.valueCounts().clear(), "Should not be modifiable")
+          .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    if (dataFile.nullValueCounts() != null) {
+      Assertions.assertThatThrownBy(
+              () -> dataFile.nullValueCounts().clear(), "Should not be modifiable")
+          .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    if (dataFile.nanValueCounts() != null) {
+      Assertions.assertThatThrownBy(
+              () -> dataFile.nanValueCounts().clear(), "Should not be modifiable")
+          .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    if (dataFile.upperBounds() != null) {
+      Assertions.assertThatThrownBy(
+              () -> dataFile.upperBounds().clear(), "Should not be modifiable")
+          .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    if (dataFile.lowerBounds() != null) {
+      Assertions.assertThatThrownBy(
+              () -> dataFile.lowerBounds().clear(), "Should not be modifiable")
+          .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    if (dataFile.columnSizes() != null) {
+      Assertions.assertThatThrownBy(
+              () -> dataFile.columnSizes().clear(), "Should not be modifiable")
+          .isInstanceOf(UnsupportedOperationException.class);
+    }
+
     Assert.assertEquals(FILE_PATH, dataFile.path()); // always select file path in all test cases
   }