Data: Add partition stats writer and reader (#11216)

Author: ajantha-bhat

api/src/main/java/org/apache/iceberg/UpdatePartitionStatistics.java

@@ -24,7 +24,7 @@
 public interface UpdatePartitionStatistics extends PendingUpdate<List<PartitionStatisticsFile>> {
   /**
    * Set the table's partition statistics file for given snapshot, replacing the previous partition
-   * statistics file for the snapshot if any exists.
+   * statistics file for the snapshot if any exists. No-op if the provided file is null.
    *
    * @return this for method chaining
    */

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

@@ -21,7 +21,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 
@@ -36,7 +35,10 @@ public SetPartitionStatistics(TableOperations ops) {
 
   @Override
   public UpdatePartitionStatistics setPartitionStatistics(PartitionStatisticsFile file) {
-    Preconditions.checkArgument(null != file, "partition statistics file must not be null");
+    if (file == null) {
+      return this;
+    }
+
     statsToSet.put(file.snapshotId(), file);
     return this;
   }

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

@@ -316,8 +316,9 @@ public void testEnableRowLineageViaPropertyAtTableCreation() {
             tableDir,
             "test",
             TEST_SCHEMA,
-            ImmutableMap.of(TableProperties.ROW_LINEAGE, "true"),
-            formatVersion);
+            PartitionSpec.unpartitioned(),
+            formatVersion,
+            ImmutableMap.of(TableProperties.ROW_LINEAGE, "true"));
     assertThat(table.ops().current().rowLineageEnabled()).isTrue();
   }
 

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

@@ -53,23 +53,13 @@ public static TestTable create(
   }
 
   public static TestTable create(
-      File temp, String name, Schema schema, Map<String, String> properties, int formatVersion) {
-    TestTableOperations ops = new TestTableOperations(name, temp);
-    if (ops.current() != null) {
-      throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
-    }
-
-    ops.commit(
-        null,
-        newTableMetadata(
-            schema,
-            PartitionSpec.unpartitioned(),
-            SortOrder.unsorted(),
-            temp.toString(),
-            properties,
-            formatVersion));
-
-    return new TestTable(ops);
+      File temp,
+      String name,
+      Schema schema,
+      PartitionSpec spec,
+      SortOrder sortOrder,
+      int formatVersion) {
+    return createTable(temp, name, schema, spec, formatVersion, ImmutableMap.of(), sortOrder, null);
   }
 
   public static TestTable create(
@@ -78,27 +68,31 @@ public static TestTable create(
       Schema schema,
       PartitionSpec spec,
       SortOrder sortOrder,
-      int formatVersion) {
-    TestTableOperations ops = new TestTableOperations(name, temp);
-    if (ops.current() != null) {
-      throw new AlreadyExistsException("Table %s already exists at location: %s", name, temp);
-    }
-
-    ops.commit(
-        null,
-        newTableMetadata(
-            schema, spec, sortOrder, temp.toString(), ImmutableMap.of(), formatVersion));
-
-    return new TestTable(ops);
+      int formatVersion,
+      MetricsReporter reporter) {
+    return createTable(
+        temp, name, schema, spec, formatVersion, ImmutableMap.of(), sortOrder, reporter);
   }
 
   public static TestTable create(
       File temp,
       String name,
       Schema schema,
       PartitionSpec spec,
-      SortOrder sortOrder,
       int formatVersion,
+      Map<String, String> properties) {
+    return createTable(
+        temp, name, schema, spec, formatVersion, properties, SortOrder.unsorted(), null);
+  }
+
+  private static TestTable createTable(
+      File temp,
+      String name,
+      Schema schema,
+      PartitionSpec spec,
+      int formatVersion,
+      Map<String, String> properties,
+      SortOrder sortOrder,
       MetricsReporter reporter) {
     TestTableOperations ops = new TestTableOperations(name, temp);
     if (ops.current() != null) {
@@ -107,10 +101,13 @@ public static TestTable create(
 
     ops.commit(
         null,
-        newTableMetadata(
-            schema, spec, sortOrder, temp.toString(), ImmutableMap.of(), formatVersion));
+        newTableMetadata(schema, spec, sortOrder, temp.toString(), properties, formatVersion));
 
-    return new TestTable(ops, reporter);
+    if (reporter != null) {
+      return new TestTable(ops, reporter);
+    } else {
+      return new TestTable(ops);
+    }
   }
 
   public static Transaction beginCreate(File temp, String name, Schema schema, PartitionSpec spec) {