Core, Data: File Format API interfaces

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

@@ -20,30 +20,22 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Map;
 import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.Schema;
 
 /**
- * Interface which should be implemented by the data file format implementations. The {@link
- * AppenderBuilder} will be parametrized based on the user provided configuration and finally the
- * {@link AppenderBuilder#build(AppenderBuilder.WriteMode)} method is used to generate the appender
- * for the specific writer use-cases. The following input should be handled by the appender in the
- * specific modes:
+ * Interface which is implemented by the data file format implementations. The {@link ObjectModel}
+ * provides the {@link AppenderBuilder} for the given parameters:
  *
  * <ul>
- *   <li>The appender's engine specific input type
- *       <ul>
- *         <li>{@link AppenderBuilder.WriteMode#DATA_WRITER}
- *         <li>{@link AppenderBuilder.WriteMode#EQUALITY_DELETE_WRITER}
- *       </ul>
- *   <li>{@link org.apache.iceberg.deletes.PositionDelete} where the type of the row is the
- *       appender's engine specific input type
- *       <ul>
- *         <li>{@link AppenderBuilder.WriteMode#POSITION_DELETE_WRITER}
- *         <li>{@link AppenderBuilder.WriteMode#POSITION_DELETE_WITH_ROW_WRITER}
- *       </ul>
+ *   <li>file format
+ *   <li>engine specific object model
+ *   <li>{@link ObjectModel.WriteMode}
  * </ul>
  *
+ * The {@link AppenderBuilder} is used to write data to the target files.
+ *
  * @param <B> type returned by builder API to allow chained calls
  * @param <E> the engine specific schema of the input data
  */
@@ -60,6 +52,11 @@ public interface AppenderBuilder<B extends AppenderBuilder<B, E>, E> {
    */
   B set(String property, String value);
 
+  default B set(Map<String, String> properties) {
+    properties.forEach(this::set);
+    return (B) this;
+  }
+
   /**
    * Set a file metadata property in the created file.
    *
@@ -107,30 +104,6 @@ default B aadPrefix(ByteBuffer aadPrefix) {
    */
   B engineSchema(E newEngineSchema);
 
-  /**
-   * Builds the {@link FileAppender} for the configured {@link WriteMode}. Could change several
-   * use-case specific configurations, like:
-   *
-   * <ul>
-   *   <li>Mode specific writer context (typically different for data and delete files).
-   *   <li>Writer functions to accept data rows, or {@link
-   *       org.apache.iceberg.deletes.PositionDelete}s
-   * </ul>
-   */
-  <D> FileAppender<D> build(WriteMode mode) throws IOException;
-
-  /**
-   * Writer modes. Based on the mode {@link #build(WriteMode)} could alter the appender
-   * configuration when creating the {@link FileAppender}.
-   */
-  enum WriteMode {
-    /** Mode for writing data files. */
-    DATA_WRITER,
-    /** Mode for writing equality delete files. */
-    EQUALITY_DELETE_WRITER,
-    /** Mode for writing position delete files. */
-    POSITION_DELETE_WRITER,
-    /** Mode for writing position delete files with row data. */
-    POSITION_DELETE_WITH_ROW_WRITER
-  }
+  /** Finalizes the configuration and builds the {@link FileAppender}. */
+  <D> FileAppender<D> build() throws IOException;
 }

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

@@ -59,13 +59,27 @@ public interface ObjectModel<E> {
 
   /**
    * The appender builder for the output file which writes the data in the specified file format and
-   * accepts the records defined by this object model.
+   * accepts the records defined by this object model. The 'mode' parameter defines the input type
+   * for the specific writer use-cases. The appender should handle the following input in the
+   * specific modes:
+   *
+   * <ul>
+   *   <li>The appender's engine specific input type
+   *       <ul>
+   *         <li>{@link WriteMode#DATA_WRITER}
+   *         <li>{@link WriteMode#EQUALITY_DELETE_WRITER}
+   *       </ul>
+   *   <li>{@link org.apache.iceberg.deletes.PositionDelete} where the type of the row is the
+   *       appender's engine specific input type when the 'mode' is {@link
+   *       WriteMode#POSITION_DELETE_WRITER}
+   * </ul>
    *
    * @param outputFile to write to
+   * @param mode for the appender
    * @return the appender builder
    * @param <B> The type of the appender builder
    */
-  <B extends AppenderBuilder<B, E>> B appenderBuilder(OutputFile outputFile);
+  <B extends AppenderBuilder<B, E>> B appenderBuilder(OutputFile outputFile, WriteMode mode);
 
   /**
    * The reader builder for the input file which reads the data from the specified file format and
@@ -76,4 +90,17 @@ public interface ObjectModel<E> {
    * @param <B> The type of the reader builder
    */
   <B extends ReadBuilder<B>> B readBuilder(InputFile inputFile);
+
+  /**
+   * Writer modes. Based on the mode the object model could alter the appender configuration when
+   * creating the {@link FileAppender}.
+   */
+  enum WriteMode {
+    /** Mode for writing data files. */
+    DATA_WRITER,
+    /** Mode for writing equality delete files. */
+    EQUALITY_DELETE_WRITER,
+    /** Mode for writing position delete files. */
+    POSITION_DELETE_WRITER,
+  }
 }

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

@@ -19,7 +19,7 @@
 package org.apache.iceberg.data;
 
 import java.util.List;
-import java.util.concurrent.ConcurrentMap;
+import java.util.Map;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.common.DynMethods;
 import org.apache.iceberg.encryption.EncryptedOutputFile;
@@ -36,10 +36,10 @@
 
 /**
  * Registry which provides the available {@link ReadBuilder}s and writer builders ({@link
- * org.apache.iceberg.data.AppenderBuilder}, {@link DataWriterBuilder}, {@link
- * EqualityDeleteWriterBuilder}, {@link PositionDeleteWriterBuilder}). Based on the `file format`
- * and the requested `object model name` the registry returns the correct reader and writer
- * builders. These builders could be used to generate the readers and writers.
+ * AppenderBuilder}, {@link DataWriterBuilder}, {@link EqualityDeleteWriterBuilder}, {@link
+ * PositionDeleteWriterBuilder}). Based on the `file format` and the requested `object model name`
+ * the registry returns the correct reader and writer builders. These builders could be used to
+ * generate the readers and writers.
  *
  * <p>The available {@link ObjectModel}s are registered by the {@link
  * #registerObjectModel(ObjectModel)} method. These {@link ObjectModel}s will be used to create the
@@ -49,9 +49,13 @@
 public final class ObjectModelRegistry {
   private static final Logger LOG = LoggerFactory.getLogger(ObjectModelRegistry.class);
   // The list of classes which are used for registering the reader and writer builders
-  private static final List<String> CLASSES_TO_REGISTER = ImmutableList.of();
+  private static final List<String> CLASSES_TO_REGISTER =
+      ImmutableList.of(
+          "org.apache.iceberg.arrow.vectorized.ArrowReader",
+          "org.apache.iceberg.flink.data.FlinkObjectModels",
+          "org.apache.iceberg.spark.source.SparkObjectModels");
 
-  private static final ConcurrentMap<Key, ObjectModel<?>> OBJECT_MODELS = Maps.newConcurrentMap();
+  private static final Map<Key, ObjectModel<?>> OBJECT_MODELS = Maps.newConcurrentMap();
 
   /**
    * Registers a new object model.
@@ -116,9 +120,10 @@ public static ReadBuilder<?> readBuilder(
    * @param <E> type for the engine specific schema expected by the appender
    * @return {@link ReadBuilder} for building the actual reader
    */
-  public static <E> org.apache.iceberg.data.AppenderBuilder<?, E> appenderBuilder(
+  public static <E> AppenderBuilder<?, E> appenderBuilder(
       FileFormat format, String objectModelName, EncryptedOutputFile outputFile) {
-    return writerFor(format, objectModelName, outputFile);
+    return ((ObjectModel<E>) OBJECT_MODELS.get(new Key(format, objectModelName)))
+        .appenderBuilder(outputFile.encryptingOutputFile(), ObjectModel.WriteMode.DATA_WRITER);
   }
 
   /**
@@ -133,7 +138,7 @@ public static <E> org.apache.iceberg.data.AppenderBuilder<?, E> appenderBuilder(
    */
   public static <E> DataWriterBuilder<?, E> writerBuilder(
       FileFormat format, String objectModelName, EncryptedOutputFile outputFile) {
-    return writerFor(format, objectModelName, outputFile);
+    return writerFor(format, objectModelName, outputFile, ObjectModel.WriteMode.DATA_WRITER);
   }
 
   /**
@@ -148,7 +153,8 @@ public static <E> DataWriterBuilder<?, E> writerBuilder(
    */
   public static <E> EqualityDeleteWriterBuilder<?, E> equalityDeleteWriterBuilder(
       FileFormat format, String objectModelName, EncryptedOutputFile outputFile) {
-    return writerFor(format, objectModelName, outputFile);
+    return writerFor(
+        format, objectModelName, outputFile, ObjectModel.WriteMode.EQUALITY_DELETE_WRITER);
   }
 
   /**
@@ -164,15 +170,19 @@ public static <E> EqualityDeleteWriterBuilder<?, E> equalityDeleteWriterBuilder(
    */
   public static <E> PositionDeleteWriterBuilder<?, E> positionDeleteWriterBuilder(
       FileFormat format, String objectModelName, EncryptedOutputFile outputFile) {
-    return writerFor(format, objectModelName, outputFile);
+    return writerFor(
+        format, objectModelName, outputFile, ObjectModel.WriteMode.POSITION_DELETE_WRITER);
   }
 
   @SuppressWarnings("unchecked")
   private static <B extends AppenderBuilder<B, E>, E> WriteBuilder<?, ?, E> writerFor(
-      FileFormat format, String objectModelName, EncryptedOutputFile outputFile) {
+      FileFormat format,
+      String objectModelName,
+      EncryptedOutputFile outputFile,
+      ObjectModel.WriteMode mode) {
     return new WriteBuilder<>(
         ((ObjectModel<E>) OBJECT_MODELS.get(new Key(format, objectModelName)))
-            .<B>appenderBuilder(outputFile.encryptingOutputFile()),
+            .<B>appenderBuilder(outputFile.encryptingOutputFile(), mode),
         outputFile.encryptingOutputFile().location(),
         format);
   }

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

@@ -54,16 +54,15 @@
  *
  * The builder wraps the file format specific {@link AppenderBuilder}. To allow further engine and
  * file format specific configuration changes for the given writer the {@link
- * AppenderBuilder#build(AppenderBuilder.WriteMode)} method is called with the correct parameter to
- * create the appender used internally to provide the required functionality.
+ * AppenderBuilder#build()} method is called to create the appender used internally to provide the
+ * required functionality.
  *
  * @param <A> type of the appender
  * @param <E> engine specific schema of the input records used for appender initialization
  */
 @SuppressWarnings("unchecked")
 class WriteBuilder<B extends WriteBuilder<B, A, E>, A extends AppenderBuilder<A, E>, E>
-    implements org.apache.iceberg.data.AppenderBuilder<B, E>,
-        DataWriterBuilder<B, E>,
+    implements DataWriterBuilder<B, E>,
         EqualityDeleteWriterBuilder<B, E>,
         PositionDeleteWriterBuilder<B, E> {
   private final AppenderBuilder<A, E> appenderBuilder;
@@ -184,11 +183,6 @@ public B withSortOrder(SortOrder newSortOrder) {
     return (B) this;
   }
 
-  @Override
-  public <D> FileAppender<D> appender() throws IOException {
-    return appenderBuilder.build(AppenderBuilder.WriteMode.DATA_WRITER);
-  }
-
   @Override
   public <D> DataWriter<D> dataWriter() throws IOException {
     Preconditions.checkArgument(spec != null, "Cannot create data writer without spec");
@@ -197,13 +191,7 @@ public <D> DataWriter<D> dataWriter() throws IOException {
         "Partition must not be null when creating data writer for partitioned spec");
 
     return new DataWriter<>(
-        appenderBuilder.build(AppenderBuilder.WriteMode.DATA_WRITER),
-        format,
-        location,
-        spec,
-        partition,
-        keyMetadata,
-        sortOrder);
+        appenderBuilder.build(), format, location, spec, partition, keyMetadata, sortOrder);
   }
 
   @Override
@@ -227,7 +215,7 @@ public <D> EqualityDeleteWriter<D> equalityDeleteWriter() throws IOException {
                 IntStream.of(equalityFieldIds)
                     .mapToObj(Objects::toString)
                     .collect(Collectors.joining(", ")))
-            .build(AppenderBuilder.WriteMode.EQUALITY_DELETE_WRITER),
+            .build(),
         format,
         location,
         spec,
@@ -251,10 +239,7 @@ public <D> PositionDeleteWriter<D> positionDeleteWriter() throws IOException {
         appenderBuilder
             .meta("delete-type", "position")
             .schema(DeleteSchemaUtil.posDeleteSchema(rowSchema))
-            .build(
-                rowSchema != null
-                    ? AppenderBuilder.WriteMode.POSITION_DELETE_WITH_ROW_WRITER
-                    : AppenderBuilder.WriteMode.POSITION_DELETE_WRITER),
+            .build(),
         format,
         location,
         spec,

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

@@ -47,8 +47,8 @@ interface WriterBuilderBase<B extends WriterBuilderBase<B, E>, E> {
   B schema(Schema newSchema);
 
   /**
-   * Sets the engine specific schema for the input. Used by the {@link
-   * AppenderBuilder#build(AppenderBuilder.WriteMode)} to configure the engine specific converters.
+   * Sets the engine specific schema for the input. Used by the {@link AppenderBuilder#build()} to
+   * configure the engine specific converters.
    */
   B engineSchema(E engineSchema);