add tests

Author: xxubai

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

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.Map;
+import java.util.function.Supplier;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.PartitionSpec;
@@ -53,14 +54,16 @@ public class GenericAppenderFactory implements FileAppenderFactory<Record> {
   private final Schema posDeleteRowSchema;
   private final Map<String, String> config = Maps.newHashMap();
 
+  private static final String WRITE_METRICS_PREFIX = "write.metadata.metrics.";
+
   @Deprecated
   public GenericAppenderFactory(Schema schema) {
-    this(null, schema, PartitionSpec.unpartitioned(), null, null, null);
+    this(schema, PartitionSpec.unpartitioned());
   }
 
   @Deprecated
   public GenericAppenderFactory(Schema schema, PartitionSpec spec) {
-    this(null, schema, spec, null, null, null);
+    this(schema, spec, null, null, null);
   }
 
   @Deprecated
@@ -70,17 +73,18 @@ public GenericAppenderFactory(
       int[] equalityFieldIds,
       Schema eqDeleteRowSchema,
       Schema posDeleteRowSchema) {
-    this(null, schema, spec, equalityFieldIds, eqDeleteRowSchema, posDeleteRowSchema);
+    this(null, schema, spec, null, equalityFieldIds, eqDeleteRowSchema, posDeleteRowSchema);
   }
 
   public GenericAppenderFactory(Table table) {
-    this(table, null, null, null, null, null);
+    this(table, null, null, null, null, null, null);
   }
 
   public GenericAppenderFactory(
       Table table,
       Schema schema,
       PartitionSpec spec,
+      Map<String, String> config,
       int[] equalityFieldIds,
       Schema eqDeleteRowSchema,
       Schema posDeleteRowSchema) {
@@ -97,17 +101,33 @@ public GenericAppenderFactory(
       this.spec = spec;
     }
 
+    if (config != null) {
+      this.config.putAll(config);
+    }
+
     this.equalityFieldIds = equalityFieldIds;
     this.eqDeleteRowSchema = eqDeleteRowSchema;
     this.posDeleteRowSchema = posDeleteRowSchema;
   }
 
   public GenericAppenderFactory set(String property, String value) {
+    if (property.startsWith(WRITE_METRICS_PREFIX) && table != null) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot set metrics property: %s directly. Use table properties instead.", property));
+    }
+
     config.put(property, value);
     return this;
   }
 
   public GenericAppenderFactory setAll(Map<String, String> properties) {
+    if (properties.keySet().stream().anyMatch(k -> k.startsWith(WRITE_METRICS_PREFIX))
+        && table != null) {
+      throw new IllegalArgumentException(
+          "Cannot set metrics properties directly. Use table properties instead.");
+    }
+
     config.putAll(properties);
     return this;
   }
@@ -120,7 +140,7 @@ public FileAppender<Record> newAppender(OutputFile outputFile, FileFormat fileFo
   @Override
   public FileAppender<Record> newAppender(
       EncryptedOutputFile encryptedOutputFile, FileFormat fileFormat) {
-    MetricsConfig metricsConfig = metricsConfig();
+    MetricsConfig metricsConfig = applyMetricsConfig(() -> MetricsConfig.forTable(table));
 
     try {
       switch (fileFormat) {
@@ -181,7 +201,7 @@ public EqualityDeleteWriter<Record> newEqDeleteWriter(
     Preconditions.checkNotNull(
         eqDeleteRowSchema,
         "Equality delete row schema shouldn't be null when creating equality-delete writer");
-    MetricsConfig metricsConfig = metricsConfig();
+    MetricsConfig metricsConfig = applyMetricsConfig(() -> MetricsConfig.forTable(table));
 
     try {
       switch (format) {
@@ -235,7 +255,7 @@ public EqualityDeleteWriter<Record> newEqDeleteWriter(
   @Override
   public PositionDeleteWriter<Record> newPosDeleteWriter(
       EncryptedOutputFile file, FileFormat format, StructLike partition) {
-    MetricsConfig metricsConfig = metricsConfig();
+    MetricsConfig metricsConfig = applyMetricsConfig(() -> MetricsConfig.forPositionDelete(table));
 
     try {
       switch (format) {
@@ -282,12 +302,12 @@ public PositionDeleteWriter<Record> newPosDeleteWriter(
     }
   }
 
-  private MetricsConfig metricsConfig() {
+  private MetricsConfig applyMetricsConfig(Supplier<MetricsConfig> metricsConfigSupplier) {
     MetricsConfig metricsConfig;
     if (table == null) {
       metricsConfig = MetricsConfig.fromProperties(config);
     } else {
-      metricsConfig = MetricsConfig.forTable(table);
+      metricsConfig = metricsConfigSupplier.get();
     }
 
     return metricsConfig;

data/src/test/java/org/apache/iceberg/TestGenericAppenderFactory.java

@@ -18,15 +18,21 @@
  */
 package org.apache.iceberg;
 
+import static org.assertj.core.api.Assertions.assertThatNoException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
 import java.util.List;
+import java.util.Map;
 import org.apache.iceberg.data.GenericAppenderFactory;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.io.FileAppenderFactory;
 import org.apache.iceberg.io.TestAppenderFactory;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.ArrayUtil;
 import org.apache.iceberg.util.StructLikeSet;
+import org.junit.jupiter.api.TestTemplate;
 
 public class TestGenericAppenderFactory extends TestAppenderFactory<Record> {
 
@@ -36,8 +42,10 @@ public class TestGenericAppenderFactory extends TestAppenderFactory<Record> {
   protected FileAppenderFactory<Record> createAppenderFactory(
       List<Integer> equalityFieldIds, Schema eqDeleteSchema, Schema posDeleteRowSchema) {
     return new GenericAppenderFactory(
+        table,
         table.schema(),
         table.spec(),
+        Maps.newHashMap(),
         ArrayUtil.toIntArray(equalityFieldIds),
         eqDeleteSchema,
         posDeleteRowSchema);
@@ -54,4 +62,60 @@ protected StructLikeSet expectedRowSet(Iterable<Record> records) {
     records.forEach(set::add);
     return set;
   }
+
+  @TestTemplate
+  void illegalSetConfig() {
+    GenericAppenderFactory appenderFactory =
+        (GenericAppenderFactory) createAppenderFactory(null, null, null);
+
+    assertThatThrownBy(
+            () ->
+                appenderFactory.set(
+                    TableProperties.METRICS_MAX_INFERRED_COLUMN_DEFAULTS,
+                    MetricsModes.None.get().toString()))
+        .as("Should not allow setting metrics property if the table was provided")
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining(
+            "Cannot set metrics property: " + TableProperties.METRICS_MAX_INFERRED_COLUMN_DEFAULTS);
+  }
+
+  @TestTemplate
+  void illegalSetAllConfigs() {
+    GenericAppenderFactory appenderFactory =
+        (GenericAppenderFactory) createAppenderFactory(null, null, null);
+
+    Map<String, String> properties =
+        ImmutableMap.of(
+            TableProperties.METRICS_MAX_INFERRED_COLUMN_DEFAULTS,
+            "10",
+            TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id",
+            MetricsModes.Full.get().toString());
+
+    assertThatThrownBy(() -> appenderFactory.setAll(properties))
+        .as("Should not allow setting metrics property if the table was provided")
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Cannot set metrics properties directly");
+  }
+
+  @TestTemplate
+  void setConfigExcludeMetrics() {
+    GenericAppenderFactory appenderFactory =
+        (GenericAppenderFactory) createAppenderFactory(null, null, null);
+    assertThatNoException().isThrownBy(() -> appenderFactory.set("key1", "value1"));
+    assertThatNoException()
+        .isThrownBy(() -> appenderFactory.setAll(ImmutableMap.of("key2", "value2")));
+  }
+
+  @TestTemplate
+  void setConfigWithoutTable() {
+    GenericAppenderFactory appenderFactory = new GenericAppenderFactory(SCHEMA);
+    assertThatNoException()
+        .isThrownBy(
+            () -> appenderFactory.set(TableProperties.METRICS_MAX_INFERRED_COLUMN_DEFAULTS, "10"));
+    assertThatNoException()
+        .isThrownBy(
+            () ->
+                appenderFactory.setAll(
+                    ImmutableMap.of(TableProperties.DEFAULT_WRITE_METRICS_MODE, "full")));
+  }
 }