Ryan's comments

Author: pvary

core/src/main/java/org/apache/iceberg/deletes/PositionDelete.java

@@ -31,6 +31,11 @@ public static <T> PositionDelete<T> create() {
 
   private PositionDelete() {}
 
+  @SuppressWarnings("unchecked")
+  public static <T> Class<PositionDelete<T>> deleteClass() {
+    return (Class<PositionDelete<T>>) (Class<?>) PositionDelete.class;
+  }
+
   public PositionDelete<R> set(CharSequence newPath, long newPos) {
     this.path = newPath;
     this.pos = newPos;

core/src/main/java/org/apache/iceberg/formats/FormatModelRegistry.java

@@ -170,12 +170,10 @@ public static <D, S> FileWriterBuilder<EqualityDeleteWriter<D>, S> equalityDelet
    * @param outputFile destination for the written data
    * @return a configured delete write builder for creating a {@link PositionDeleteWriter}
    */
-  @SuppressWarnings("unchecked")
   public static <D> FileWriterBuilder<PositionDeleteWriter<D>, ?> positionDeleteWriteBuilder(
       FileFormat format, EncryptedOutputFile outputFile) {
-    Class<PositionDelete<D>> deleteClass =
-        (Class<PositionDelete<D>>) (Class<?>) PositionDelete.class;
-    FormatModel<PositionDelete<D>, ?> model = FormatModelRegistry.modelFor(format, deleteClass);
+    FormatModel<PositionDelete<D>, ?> model =
+        FormatModelRegistry.modelFor(format, PositionDelete.deleteClass());
     return FileWriterBuilderImpl.forPositionDelete(model, outputFile);
   }
 

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

@@ -18,7 +18,6 @@
  */
 package org.apache.iceberg.data;
 
-import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.parquet.GenericParquetReaders;
 import org.apache.iceberg.data.parquet.GenericParquetWriter;
 import org.apache.iceberg.formats.FormatModelRegistry;
@@ -29,13 +28,13 @@ public static void register() {
     FormatModelRegistry.register(
         ParquetFormatModel.create(
             Record.class,
-            Schema.class,
+            Void.class,
             (icebergSchema, fileSchema, engineSchema) ->
                 GenericParquetWriter.create(icebergSchema, fileSchema),
             (icebergSchema, fileSchema, engineSchema, idToConstant) ->
                 GenericParquetReaders.buildReader(icebergSchema, fileSchema, idToConstant)));
 
-    FormatModelRegistry.register(ParquetFormatModel.forDelete());
+    FormatModelRegistry.register(ParquetFormatModel.forPositionDeletes());
   }
 
   private GenericFormatModels() {}

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

@@ -31,6 +31,14 @@
 public class DataTestHelpers {
   private DataTestHelpers() {}
 
+  public static void assertEquals(
+      Types.StructType struct, List<Record> expected, List<Record> actual) {
+    assertThat(actual).hasSize(expected.size());
+    for (int i = 0; i < expected.size(); i += 1) {
+      assertEquals(struct, expected.get(i), actual.get(i), null, -1);
+    }
+  }
+
   public static void assertEquals(Types.StructType struct, Record expected, Record actual) {
     assertEquals(struct, expected, actual, null, -1);
   }

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

@@ -28,9 +28,6 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileFormat;
-import org.apache.iceberg.Parameter;
-import org.apache.iceberg.ParameterizedTestExtension;
-import org.apache.iceberg.Parameters;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.TestBase;
@@ -46,28 +43,18 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.DataWriter;
 import org.apache.iceberg.io.InputFile;
-import org.apache.iceberg.io.OutputFile;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.TestTemplate;
-import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
 
-@ExtendWith(ParameterizedTestExtension.class)
 public class TestGenericFormatModels {
-  @Parameters(name = "fileFormat = {0}")
-  protected static List<Object> parameters() {
-    return List.of(FileFormat.PARQUET);
-  }
-
   private static final List<Record> TEST_RECORDS =
-      ImmutableList.of(
-          GenericRecord.create(TestBase.SCHEMA).copy("id", 1, "data", "hello"),
-          GenericRecord.create(TestBase.SCHEMA).copy("id", 2, "data", "world"));
+      RandomGenericData.generate(TestBase.SCHEMA, 10, 1L);
 
-  @Parameter(index = 0)
-  private FileFormat fileFormat;
+  private static final FileFormat[] FILE_FORMATS = new FileFormat[] {FileFormat.PARQUET};
 
   @TempDir protected Path temp;
 
@@ -77,8 +64,9 @@ protected static List<Object> parameters() {
   @BeforeEach
   public void before() {
     this.fileIO = new InMemoryFileIO();
-    OutputFile outputFile = fileIO.newOutputFile("test-file." + fileFormat.name().toLowerCase());
-    this.encryptedFile = EncryptedFiles.encryptedOutput(outputFile, EncryptionKeyMetadata.EMPTY);
+    this.encryptedFile =
+        EncryptedFiles.encryptedOutput(
+            fileIO.newOutputFile("test-file"), EncryptionKeyMetadata.EMPTY);
   }
 
   @AfterEach
@@ -90,8 +78,9 @@ public void after() throws IOException {
     }
   }
 
-  @TestTemplate
-  public void testDataWriter() throws IOException {
+  @ParameterizedTest
+  @FieldSource("FILE_FORMATS")
+  public void testDataWriter(FileFormat fileFormat) throws IOException {
     FileWriterBuilder<DataWriter<Record>, Schema> writerBuilder =
         FormatModelRegistry.dataWriteBuilder(fileFormat, Record.class, encryptedFile);
 
@@ -107,11 +96,11 @@ public void testDataWriter() throws IOException {
     dataFile = writer.toDataFile();
 
     assertThat(dataFile).isNotNull();
-    assertThat(dataFile.recordCount()).isEqualTo(2);
+    assertThat(dataFile.recordCount()).isEqualTo(TEST_RECORDS.size());
     assertThat(dataFile.format()).isEqualTo(fileFormat);
 
     // Verify the file content by reading it back
-    InputFile inputFile = fileIO.newInputFile(encryptedFile.encryptingOutputFile().location());
+    InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
     List<Record> readRecords;
     try (CloseableIterable<Record> reader =
         FormatModelRegistry.readBuilder(fileFormat, Record.class, inputFile)
@@ -120,15 +109,12 @@ public void testDataWriter() throws IOException {
       readRecords = ImmutableList.copyOf(reader);
     }
 
-    assertThat(readRecords).hasSize(2);
-    DataTestHelpers.assertEquals(
-        TestBase.SCHEMA.asStruct(), TEST_RECORDS.get(0), readRecords.get(0));
-    DataTestHelpers.assertEquals(
-        TestBase.SCHEMA.asStruct(), TEST_RECORDS.get(1), readRecords.get(1));
+    DataTestHelpers.assertEquals(TestBase.SCHEMA.asStruct(), TEST_RECORDS, readRecords);
   }
 
-  @TestTemplate
-  public void testEqualityDeleteWriter() throws IOException {
+  @ParameterizedTest
+  @FieldSource("FILE_FORMATS")
+  public void testEqualityDeleteWriter(FileFormat fileFormat) throws IOException {
     FileWriterBuilder<EqualityDeleteWriter<Record>, Schema> writerBuilder =
         FormatModelRegistry.equalityDeleteWriteBuilder(fileFormat, Record.class, encryptedFile);
 
@@ -148,12 +134,12 @@ public void testEqualityDeleteWriter() throws IOException {
     deleteFile = writer.toDeleteFile();
 
     assertThat(deleteFile).isNotNull();
-    assertThat(deleteFile.recordCount()).isEqualTo(2);
+    assertThat(deleteFile.recordCount()).isEqualTo(TEST_RECORDS.size());
     assertThat(deleteFile.format()).isEqualTo(fileFormat);
     assertThat(deleteFile.equalityFieldIds()).containsExactly(3);
 
     // Verify the file content by reading it back
-    InputFile inputFile = fileIO.newInputFile(encryptedFile.encryptingOutputFile().location());
+    InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
     List<Record> readRecords;
     try (CloseableIterable<Record> reader =
         FormatModelRegistry.readBuilder(fileFormat, Record.class, inputFile)
@@ -162,25 +148,22 @@ public void testEqualityDeleteWriter() throws IOException {
       readRecords = ImmutableList.copyOf(reader);
     }
 
-    assertThat(readRecords).hasSize(2);
-    DataTestHelpers.assertEquals(
-        TestBase.SCHEMA.asStruct(), TEST_RECORDS.get(0), readRecords.get(0));
-    DataTestHelpers.assertEquals(
-        TestBase.SCHEMA.asStruct(), TEST_RECORDS.get(1), readRecords.get(1));
+    DataTestHelpers.assertEquals(TestBase.SCHEMA.asStruct(), TEST_RECORDS, readRecords);
   }
 
-  @TestTemplate
-  public void testPositionDeleteWriter() throws IOException {
+  @ParameterizedTest
+  @FieldSource("FILE_FORMATS")
+  public void testPositionDeleteWriter(FileFormat fileFormat) throws IOException {
     Schema positionDeleteSchema = new Schema(DELETE_FILE_PATH, DELETE_FILE_POS);
 
     FileWriterBuilder<PositionDeleteWriter<Record>, ?> writerBuilder =
         FormatModelRegistry.positionDeleteWriteBuilder(fileFormat, encryptedFile);
 
     PositionDelete<Record> delete1 = PositionDelete.create();
-    delete1.set("data-file-1.parquet", 0L, null);
+    delete1.set("data-file-1.parquet", 0L);
 
     PositionDelete<Record> delete2 = PositionDelete.create();
-    delete2.set("data-file-1.parquet", 1L, null);
+    delete2.set("data-file-1.parquet", 1L);
 
     List<PositionDelete<Record>> positionDeletes = ImmutableList.of(delete1, delete2);
 
@@ -199,7 +182,7 @@ public void testPositionDeleteWriter() throws IOException {
     assertThat(deleteFile.format()).isEqualTo(fileFormat);
 
     // Verify the file content by reading it back
-    InputFile inputFile = fileIO.newInputFile(encryptedFile.encryptingOutputFile().location());
+    InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
     List<Record> readRecords;
     try (CloseableIterable<Record> reader =
         FormatModelRegistry.readBuilder(fileFormat, Record.class, inputFile)
@@ -208,10 +191,13 @@ public void testPositionDeleteWriter() throws IOException {
       readRecords = ImmutableList.copyOf(reader);
     }
 
-    assertThat(readRecords).hasSize(2);
-    assertThat(readRecords.get(0).getField("file_path")).isEqualTo("data-file-1.parquet");
-    assertThat(readRecords.get(0).getField("pos")).isEqualTo(0L);
-    assertThat(readRecords.get(1).getField("file_path")).isEqualTo("data-file-1.parquet");
-    assertThat(readRecords.get(1).getField("pos")).isEqualTo(1L);
+    List<Record> expected =
+        ImmutableList.of(
+            GenericRecord.create(positionDeleteSchema)
+                .copy(DELETE_FILE_PATH.name(), "data-file-1.parquet", DELETE_FILE_POS.name(), 0L),
+            GenericRecord.create(positionDeleteSchema)
+                .copy(DELETE_FILE_PATH.name(), "data-file-1.parquet", DELETE_FILE_POS.name(), 1L));
+
+    DataTestHelpers.assertEquals(positionDeleteSchema.asStruct(), expected, readRecords);
   }
 }

parquet/src/main/java/org/apache/iceberg/parquet/ParquetFormatModel.java

@@ -38,18 +38,18 @@
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.parquet.column.ParquetProperties;
 import org.apache.parquet.schema.MessageType;
 
 public class ParquetFormatModel<D, S, R>
     extends BaseFormatModel<D, S, ParquetValueWriter<?>, R, MessageType> {
   public static final String WRITER_VERSION_KEY = "parquet.writer.version";
-  private final boolean batchReader;
+  private final boolean isBatchReader;
 
-  @SuppressWarnings("rawtypes")
-  public static ParquetFormatModel<PositionDelete, Object, Object> forDelete() {
-    return new ParquetFormatModel<>(PositionDelete.class, null, null, null, false);
+  public static <D> ParquetFormatModel<PositionDelete<D>, Object, Object> forPositionDeletes() {
+    return new ParquetFormatModel<>(PositionDelete.deleteClass(), null, null, null, false);
   }
 
   public static <D, S> ParquetFormatModel<D, S, ParquetValueReader<?>> create(
@@ -72,9 +72,9 @@ private ParquetFormatModel(
       Class<S> schemaType,
       WriterFunction<ParquetValueWriter<?>, S, MessageType> writerFunction,
       ReaderFunction<R, S, MessageType> readerFunction,
-      boolean batchReader) {
+      boolean isBatchReader) {
     super(type, schemaType, writerFunction, readerFunction);
-    this.batchReader = batchReader;
+    this.isBatchReader = isBatchReader;
   }
 
   @Override
@@ -89,12 +89,13 @@ public ModelWriteBuilder<D, S> writeBuilder(EncryptedOutputFile outputFile) {
 
   @Override
   public ReadBuilder<D, S> readBuilder(InputFile inputFile) {
-    return new ReadBuilderWrapper<>(inputFile, readerFunction(), batchReader);
+    return new ReadBuilderWrapper<>(inputFile, readerFunction(), isBatchReader);
   }
 
   private static class WriteBuilderWrapper<D, S> implements ModelWriteBuilder<D, S> {
     private final Parquet.WriteBuilder internal;
     private final WriterFunction<ParquetValueWriter<?>, S, MessageType> writerFunction;
+    private Schema schema;
     private S engineSchema;
     private FileContent content;
 
@@ -106,14 +107,15 @@ private WriteBuilderWrapper(
     }
 
     @Override
-    public ModelWriteBuilder<D, S> schema(Schema schema) {
-      internal.schema(schema);
+    public ModelWriteBuilder<D, S> schema(Schema newSchema) {
+      this.schema = newSchema;
+      internal.schema(newSchema);
       return this;
     }
 
     @Override
-    public ModelWriteBuilder<D, S> engineSchema(S schema) {
-      this.engineSchema = schema;
+    public ModelWriteBuilder<D, S> engineSchema(S newSchema) {
+      this.engineSchema = newSchema;
       return this;
     }
 
@@ -191,6 +193,15 @@ public FileAppender<D> build() throws IOException {
                   writerFunction.write(icebergSchema, messageType, engineSchema));
           break;
         case POSITION_DELETES:
+          Preconditions.checkState(
+              schema == null,
+              "Invalid schema: %s. Position deletes with schema are not supported by the API.",
+              schema);
+          Preconditions.checkState(
+              engineSchema == null,
+              "Invalid engineSchema: %s. Position deletes with schema are not supported by the API.",
+              engineSchema);
+
           internal.createContextFunc(Parquet.WriteBuilder.Context::deleteContext);
           internal.createWriterFunc(
               (icebergSchema, messageType) ->