Core: Deprecate and remove GenericAppenderFactory from tests (#14353)

Author: pvary

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

@@ -55,6 +55,14 @@ public final class MetricsConfig implements Serializable {
   private static final MetricsMode DEFAULT_MODE =
       MetricsModes.fromString(DEFAULT_WRITE_METRICS_MODE_DEFAULT);
   private static final MetricsConfig DEFAULT = new MetricsConfig(ImmutableMap.of(), DEFAULT_MODE);
+  private static final MetricsConfig POSITION_DELETE_MODE =
+      new MetricsConfig(
+          ImmutableMap.of(
+              MetadataColumns.DELETE_FILE_PATH.name(),
+              MetricsModes.Full.get(),
+              MetadataColumns.DELETE_FILE_POS.name(),
+              MetricsModes.Full.get()),
+          DEFAULT_MODE);
 
   private final Map<String, MetricsMode> columnModes;
   private final MetricsMode defaultMode;
@@ -68,6 +76,10 @@ public static MetricsConfig getDefault() {
     return DEFAULT;
   }
 
+  public static MetricsConfig forPositionDelete() {
+    return POSITION_DELETE_MODE;
+  }
+
   /**
    * Creates a metrics config from table configuration.
    *
@@ -92,7 +104,11 @@ public static MetricsConfig forTable(Table table) {
    * Creates a metrics config for a position delete file.
    *
    * @param table an Iceberg table
+   * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0.
+   *     Position deletes that include row data are no longer supported. Use {@link
+   *     #forPositionDelete()} instead.
    */
+  @Deprecated
   public static MetricsConfig forPositionDelete(Table table) {
     ImmutableMap.Builder<String, MetricsMode> columnModes = ImmutableMap.builder();
 

core/src/main/java/org/apache/iceberg/io/SortedPosDeleteWriter.java

@@ -150,8 +150,9 @@ protected BaseFileWriterFactory(
   public DataWriter<T> newDataWriter(
       EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
     EncryptionKeyMetadata keyMetadata = file.keyMetadata();
-    Map<String, String> properties = table.properties();
-    MetricsConfig metricsConfig = MetricsConfig.forTable(table);
+    Map<String, String> properties = table == null ? ImmutableMap.of() : table.properties();
+    MetricsConfig metricsConfig =
+        table == null ? MetricsConfig.getDefault() : MetricsConfig.forTable(table);
 
     try {
       switch (dataFileFormat) {
@@ -219,8 +220,9 @@ public DataWriter<T> newDataWriter(
   public EqualityDeleteWriter<T> newEqualityDeleteWriter(
       EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
     EncryptionKeyMetadata keyMetadata = file.keyMetadata();
-    Map<String, String> properties = table.properties();
-    MetricsConfig metricsConfig = MetricsConfig.forTable(table);
+    Map<String, String> properties = table == null ? ImmutableMap.of() : table.properties();
+    MetricsConfig metricsConfig =
+        table == null ? MetricsConfig.getDefault() : MetricsConfig.forTable(table);
 
     try {
       switch (deleteFileFormat) {
@@ -291,8 +293,9 @@ public EqualityDeleteWriter<T> newEqualityDeleteWriter(
   public PositionDeleteWriter<T> newPositionDeleteWriter(
       EncryptedOutputFile file, PartitionSpec spec, StructLike partition) {
     EncryptionKeyMetadata keyMetadata = file.keyMetadata();
-    Map<String, String> properties = table.properties();
-    MetricsConfig metricsConfig = MetricsConfig.forPositionDelete(table);
+    Map<String, String> properties = table == null ? ImmutableMap.of() : table.properties();
+    MetricsConfig metricsConfig =
+        table == null ? MetricsConfig.forPositionDelete() : MetricsConfig.forPositionDelete(table);
 
     try {
       switch (deleteFileFormat) {

data/src/main/java/org/apache/iceberg/data/BaseFileWriterFactory.java

@@ -44,7 +44,12 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 
-/** Factory to create a new {@link FileAppender} to write {@link Record}s. */
+/**
+ * Factory to create a new {@link FileAppender} to write {@link Record}s.
+ *
+ * @deprecated will be removed in 1.12.0; use {@link GenericFileWriterFactory} instead.
+ */
+@Deprecated
 public class GenericAppenderFactory implements FileAppenderFactory<Record> {
   private final Table table;
   private final Schema schema;
@@ -62,12 +67,6 @@ public GenericAppenderFactory(Schema schema, PartitionSpec spec) {
     this(schema, spec, null, null, null);
   }
 
-  /**
-   * @deprecated This constructor is deprecated as of version 1.11.0 and will be removed in 1.12.0.
-   *     Position deletes that include row data are no longer supported. Use {@link
-   *     #GenericAppenderFactory(Schema, PartitionSpec, int[], Schema)} instead.
-   */
-  @Deprecated
   public GenericAppenderFactory(
       Schema schema,
       PartitionSpec spec,

data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java

@@ -177,6 +177,10 @@ public static class Builder {
     private Schema positionDeleteRowSchema;
     private Map<String, String> writerProperties = ImmutableMap.of();
 
+    public Builder() {
+      this.table = null;
+    }
+
     public Builder(Table table) {
       this.table = table;
       this.dataSchema = table.schema();

data/src/main/java/org/apache/iceberg/data/GenericFileWriterFactory.java

@@ -32,7 +32,6 @@
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.RandomGenericData;
 import org.apache.iceberg.data.Record;
-import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -105,7 +104,7 @@ public abstract class TestMergingMetrics<T> {
           MAP_FIELD_2,
           STRUCT_FIELD);
 
-  protected abstract FileAppender<T> writeAndGetAppender(List<Record> records) throws Exception;
+  protected abstract DataFile writeAndGetDataFile(List<Record> records) throws Exception;
 
   @Parameters(name = "fileFormat = {0}")
   public static List<Object> parameters() {
@@ -131,11 +130,10 @@ public void verifySingleRecordMetric() throws Exception {
         ImmutableMap.of(
             0, 0D, 1, Double.NaN, 2, 2D, 3, Double.NaN, 4, Double.NaN)); // MAP_FIELD_2 - 3
 
-    FileAppender<T> appender = writeAndGetAppender(ImmutableList.of(record));
-    Metrics metrics = appender.metrics();
-    Map<Integer, Long> nanValueCount = metrics.nanValueCounts();
-    Map<Integer, ByteBuffer> upperBounds = metrics.upperBounds();
-    Map<Integer, ByteBuffer> lowerBounds = metrics.lowerBounds();
+    DataFile dataFile = writeAndGetDataFile(ImmutableList.of(record));
+    Map<Integer, Long> nanValueCount = dataFile.nanValueCounts();
+    Map<Integer, ByteBuffer> upperBounds = dataFile.upperBounds();
+    Map<Integer, ByteBuffer> lowerBounds = dataFile.lowerBounds();
 
     assertThat(nanValueCount).hasSize(2);
     assertNaNCountMatch(1L, nanValueCount, FLOAT_FIELD);
@@ -153,28 +151,27 @@ public void verifyRandomlyGeneratedRecordsMetric() throws Exception {
     // too big of the record count will more likely to make all upper/lower bounds +/-infinity,
     // which makes the tests easier to pass
     List<Record> recordList = RandomGenericData.generate(SCHEMA, 5, 250L);
-    FileAppender<T> appender = writeAndGetAppender(recordList);
+    DataFile dataFile = writeAndGetDataFile(recordList);
 
     Map<Types.NestedField, AtomicReference<Number>> expectedUpperBounds = Maps.newHashMap();
     Map<Types.NestedField, AtomicReference<Number>> expectedLowerBounds = Maps.newHashMap();
     Map<Types.NestedField, AtomicLong> expectedNaNCount = Maps.newHashMap();
 
     populateExpectedValues(recordList, expectedUpperBounds, expectedLowerBounds, expectedNaNCount);
 
-    Metrics metrics = appender.metrics();
     expectedUpperBounds.forEach(
-        (key, value) -> assertBoundValueMatch(value.get(), metrics.upperBounds(), key));
+        (key, value) -> assertBoundValueMatch(value.get(), dataFile.upperBounds(), key));
     expectedLowerBounds.forEach(
-        (key, value) -> assertBoundValueMatch(value.get(), metrics.lowerBounds(), key));
+        (key, value) -> assertBoundValueMatch(value.get(), dataFile.lowerBounds(), key));
     expectedNaNCount.forEach(
-        (key, value) -> assertNaNCountMatch(value.get(), metrics.nanValueCounts(), key));
+        (key, value) -> assertNaNCountMatch(value.get(), dataFile.nanValueCounts(), key));
 
     SCHEMA.columns().stream()
         .filter(column -> !FIELDS_WITH_NAN_COUNT_TO_ID.containsKey(column))
         .map(Types.NestedField::fieldId)
         .forEach(
             id ->
-                assertThat(metrics.nanValueCounts().get(id))
+                assertThat(dataFile.nanValueCounts().get(id))
                     .as("NaN count for field %s should be null")
                     .isNull());
   }