Flink: Add null check to writers to prevent resurrecting null values

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

@@ -27,6 +27,7 @@
 import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.nio.file.Path;
+import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Stream;
@@ -40,6 +41,7 @@
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.DateTimeUtil;
 import org.assertj.core.api.Assumptions;
+import org.junit.Assert;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -50,6 +52,8 @@ public abstract class DataTest {
 
   protected abstract void writeAndValidate(Schema schema) throws IOException;
 
+  protected abstract void writeAndValidate(Schema schema, List<Record> data) throws IOException;
+
   protected void writeAndValidate(Schema writeSchema, Schema expectedSchema) throws IOException {
     throw new UnsupportedEncodingException(
         "Cannot run test, writeAndValidate(Schema, Schema) is not implemented");
@@ -486,4 +490,20 @@ public void testPrimitiveTypeDefaultValues(Type.PrimitiveType type, Object defau
 
     writeAndValidate(writeSchema, readSchema);
   }
+
+  @Test
+  public void testWriteNullValueForRequiredType() {
+    Schema schema =
+        new Schema(
+            required(0, "id", LongType.get()), required(1, "string", Types.StringType.get()));
+
+    GenericRecord genericRecord = GenericRecord.create(schema);
+    genericRecord.set(0, 42L);
+    genericRecord.set(1, null);
+
+    Assert.assertThrows(
+        // The actual exception depends on the implementation, e.g.
+        // NullPointerException or IllegalArgumentException.
+        Exception.class, () -> writeAndValidate(schema, List.of(genericRecord)));
+  }
 }

data/src/test/java/org/apache/iceberg/data/avro/TestGenericData.java

@@ -35,14 +35,26 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 
 public class TestGenericData extends DataTest {
+
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     writeAndValidate(schema, schema);
   }
 
+  @Override
+  protected void writeAndValidate(Schema writeSchema, List<Record> expectedData)
+      throws IOException {
+    writeAndValidate(writeSchema, writeSchema, expectedData);
+  }
+
   @Override
   protected void writeAndValidate(Schema writeSchema, Schema expectedSchema) throws IOException {
-    List<Record> expected = RandomGenericData.generate(writeSchema, 100, 0L);
+    List<Record> data = RandomGenericData.generate(writeSchema, 100, 0L);
+    writeAndValidate(writeSchema, expectedSchema, data);
+  }
+
+  protected void writeAndValidate(Schema writeSchema, Schema expectedSchema, List<Record> expected)
+      throws IOException {
 
     File testFile = File.createTempFile("junit", null, temp.toFile());
     assertThat(testFile.delete()).isTrue();

data/src/test/java/org/apache/iceberg/data/orc/TestGenericData.java

@@ -50,6 +50,7 @@
 import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
 import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
 import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+import org.junit.Ignore;
 import org.junit.jupiter.api.Test;
 
 public class TestGenericData extends DataTest {
@@ -61,6 +62,11 @@ protected void writeAndValidate(Schema schema) throws IOException {
     writeAndValidateRecords(schema, expected);
   }
 
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedData) throws IOException {
+    writeAndValidateRecords(schema, expectedData);
+  }
+
   @Test
   public void writeAndValidateRepeatingRecords() throws IOException {
     Schema structSchema =
@@ -237,4 +243,10 @@ private void writeAndValidateRecords(Schema schema, List<Record> expected) throw
       DataTestHelpers.assertEquals(schema.asStruct(), expected.get(i), rows.get(i));
     }
   }
+
+  @Override
+  @Ignore("ORC file format supports null values even for required fields")
+  public void testWriteNullValueForRequiredType() {
+    super.testWriteNullValueForRequiredType();
+  }
 }

data/src/test/java/org/apache/iceberg/data/parquet/TestGenericData.java

@@ -46,14 +46,25 @@
 import org.junit.jupiter.api.Test;
 
 public class TestGenericData extends DataTest {
+
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     writeAndValidate(schema, schema);
   }
 
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expected) throws IOException {
+    writeAndValidate(schema, schema, expected);
+  }
+
   @Override
   protected void writeAndValidate(Schema writeSchema, Schema expectedSchema) throws IOException {
-    List<Record> expected = RandomGenericData.generate(writeSchema, 100, 12228L);
+    List<Record> data = RandomGenericData.generate(writeSchema, 100, 12228L);
+    writeAndValidate(writeSchema, expectedSchema, data);
+  }
+
+  protected void writeAndValidate(Schema writeSchema, Schema expectedSchema, List<Record> expected)
+      throws IOException {
 
     File testFile = File.createTempFile("junit", null, temp.toFile());
     assertThat(testFile.delete()).isTrue();

data/src/test/java/org/apache/iceberg/data/parquet/TestParquetEncryptionWithWriteSupport.java

@@ -59,6 +59,11 @@ public class TestParquetEncryptionWithWriteSupport extends DataTest {
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     List<Record> expected = RandomGenericData.generate(schema, 100, 0L);
+    writeAndValidate(schema, expected);
+  }
+
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expected) throws IOException {
 
     File testFile = File.createTempFile("junit", null, temp.toFile());
     assertThat(testFile.delete()).isTrue();

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

@@ -498,6 +498,12 @@ private static class RowDataWriter extends ParquetValueWriters.StructWriter<RowD
 
     @Override
     protected Object get(RowData struct, int index) {
+      // Be sure to check for null values, even if the field is required. Without an explicit null
+      // check, BinaryRowData will ignore the null flag and parse random bytes as actual values.
+      // This will produce incorrect writes instead of failing with a NullPointerException.
+      if (struct.isNullAt(index)) {
+        return null;
+      }
       return fieldGetter[index].getFieldOrNull(struct);
     }
   }

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/AbstractTestFlinkAvroReaderWriter.java

@@ -67,13 +67,13 @@ public abstract class AbstractTestFlinkAvroReaderWriter extends DataTest {
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1991L);
-    writeAndValidate(schema, expectedRecords, NUM_RECORDS);
+    writeAndValidate(schema, expectedRecords);
   }
 
   protected abstract Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema);
 
-  private void writeAndValidate(Schema schema, List<Record> expectedRecords, int numRecord)
-      throws IOException {
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
     RowType flinkSchema = FlinkSchemaUtil.convert(schema);
     List<RowData> expectedRows = Lists.newArrayList(RandomRowData.convert(schema, expectedRecords));
 
@@ -93,7 +93,7 @@ private void writeAndValidate(Schema schema, List<Record> expectedRecords, int n
     try (CloseableIterable<RowData> reader = createAvroReadBuilder(recordsFile, schema).build()) {
       Iterator<Record> expected = expectedRecords.iterator();
       Iterator<RowData> rows = reader.iterator();
-      for (int i = 0; i < numRecord; i++) {
+      for (int i = 0; i < expectedRecords.size(); i++) {
         assertThat(rows).hasNext();
         TestHelpers.assertRowData(schema.asStruct(), flinkSchema, expected.next(), rows.next());
       }
@@ -120,7 +120,7 @@ private void writeAndValidate(Schema schema, List<Record> expectedRecords, int n
             .build()) {
       Iterator<RowData> expected = expectedRows.iterator();
       Iterator<Record> records = reader.iterator();
-      for (int i = 0; i < numRecord; i += 1) {
+      for (int i = 0; i < expectedRecords.size(); i += 1) {
         assertThat(records).hasNext();
         TestHelpers.assertRowData(schema.asStruct(), flinkSchema, records.next(), expected.next());
       }
@@ -177,6 +177,6 @@ public void testNumericTypes() throws IOException {
                 1643811742000L,
                 10.24d));
 
-    writeAndValidate(SCHEMA_NUM_TYPE, expected, 2);
+    writeAndValidate(SCHEMA_NUM_TYPE, expected);
   }
 }

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java

@@ -39,14 +39,20 @@
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.Ignore;
 
 public class TestFlinkOrcReaderWriter extends DataTest {
   private static final int NUM_RECORDS = 100;
 
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
-    RowType flinkSchema = FlinkSchemaUtil.convert(schema);
     List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1990L);
+    writeAndValidate(schema, expectedRecords);
+  }
+
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
+    RowType flinkSchema = FlinkSchemaUtil.convert(schema);
     List<RowData> expectedRows = Lists.newArrayList(RandomRowData.convert(schema, expectedRecords));
 
     File recordsFile = File.createTempFile("junit", null, temp.toFile());
@@ -104,4 +110,10 @@ protected void writeAndValidate(Schema schema) throws IOException {
       assertThat(records).isExhausted();
     }
   }
+
+  @Override
+  @Ignore("ORC file format supports null values even for required fields")
+  public void testWriteNullValueForRequiredType() {
+    super.testWriteNullValueForRequiredType();
+  }
 }

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReader.java

@@ -236,4 +236,9 @@ protected void writeAndValidate(Schema schema) throws IOException {
         RandomGenericData.generateFallbackRecords(schema, NUM_RECORDS, 21124, NUM_RECORDS / 20),
         schema);
   }
+
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedData) throws IOException {
+    writeAndValidate(expectedData, schema);
+  }
 }

flink/v1.18/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetWriter.java

@@ -24,7 +24,10 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.Iterator;
+import java.util.List;
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
 import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.iceberg.Files;
 import org.apache.iceberg.Schema;
@@ -33,10 +36,12 @@
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.data.parquet.GenericParquetReaders;
 import org.apache.iceberg.flink.FlinkSchemaUtil;
+import org.apache.iceberg.flink.RowDataConverter;
 import org.apache.iceberg.flink.TestHelpers;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.junit.jupiter.api.io.TempDir;
 
 public class TestFlinkParquetWriter extends DataTest {
@@ -91,4 +96,16 @@ protected void writeAndValidate(Schema schema) throws IOException {
                 schema, NUM_RECORDS, 21124, NUM_RECORDS / 20)),
         schema);
   }
+
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedData) throws IOException {
+    RowDataSerializer rowDataSerializer = new RowDataSerializer(FlinkSchemaUtil.convert(schema));
+    List<RowData> binaryRowList = Lists.newArrayList();
+    for (Record record : expectedData) {
+      RowData rowData = RowDataConverter.convert(schema, record);
+      BinaryRowData binaryRow = rowDataSerializer.toBinaryRow(rowData);
+      binaryRowList.add(binaryRow);
+    }
+    writeAndValidate(binaryRowList, schema);
+  }
 }

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkOrcWriters.java

@@ -304,6 +304,9 @@ static class RowDataWriter extends GenericOrcWriters.StructWriter<RowData> {
 
     @Override
     protected Object get(RowData struct, int index) {
+      if (struct.isNullAt(index)) {
+        return null;
+      }
       return fieldGetters.get(index).getFieldOrNull(struct);
     }
   }

flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

@@ -498,6 +498,12 @@ private static class RowDataWriter extends ParquetValueWriters.StructWriter<RowD
 
     @Override
     protected Object get(RowData struct, int index) {
+      // Be sure to check for null values, even if the field is required. Without an explicit null
+      // check, BinaryRowData will ignore the null flag and parse random bytes as actual values.
+      // This will produce incorrect writes instead of failing with a NullPointerException.
+      if (struct.isNullAt(index)) {
+        return null;
+      }
       return fieldGetter[index].getFieldOrNull(struct);
     }
   }

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/AbstractTestFlinkAvroReaderWriter.java

@@ -67,13 +67,13 @@ public abstract class AbstractTestFlinkAvroReaderWriter extends DataTest {
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1991L);
-    writeAndValidate(schema, expectedRecords, NUM_RECORDS);
+    writeAndValidate(schema, expectedRecords);
   }
 
   protected abstract Avro.ReadBuilder createAvroReadBuilder(File recordsFile, Schema schema);
 
-  private void writeAndValidate(Schema schema, List<Record> expectedRecords, int numRecord)
-      throws IOException {
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
     RowType flinkSchema = FlinkSchemaUtil.convert(schema);
     List<RowData> expectedRows = Lists.newArrayList(RandomRowData.convert(schema, expectedRecords));
 
@@ -93,7 +93,7 @@ private void writeAndValidate(Schema schema, List<Record> expectedRecords, int n
     try (CloseableIterable<RowData> reader = createAvroReadBuilder(recordsFile, schema).build()) {
       Iterator<Record> expected = expectedRecords.iterator();
       Iterator<RowData> rows = reader.iterator();
-      for (int i = 0; i < numRecord; i++) {
+      for (int i = 0; i < expectedRecords.size(); i++) {
         assertThat(rows).hasNext();
         TestHelpers.assertRowData(schema.asStruct(), flinkSchema, expected.next(), rows.next());
       }
@@ -120,7 +120,7 @@ private void writeAndValidate(Schema schema, List<Record> expectedRecords, int n
             .build()) {
       Iterator<RowData> expected = expectedRows.iterator();
       Iterator<Record> records = reader.iterator();
-      for (int i = 0; i < numRecord; i += 1) {
+      for (int i = 0; i < expectedRecords.size(); i += 1) {
         assertThat(records).hasNext();
         TestHelpers.assertRowData(schema.asStruct(), flinkSchema, records.next(), expected.next());
       }
@@ -177,6 +177,6 @@ public void testNumericTypes() throws IOException {
                 1643811742000L,
                 10.24d));
 
-    writeAndValidate(SCHEMA_NUM_TYPE, expected, 2);
+    writeAndValidate(SCHEMA_NUM_TYPE, expected);
   }
 }

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java

@@ -39,14 +39,20 @@
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.junit.Ignore;
 
 public class TestFlinkOrcReaderWriter extends DataTest {
   private static final int NUM_RECORDS = 100;
 
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
-    RowType flinkSchema = FlinkSchemaUtil.convert(schema);
     List<Record> expectedRecords = RandomGenericData.generate(schema, NUM_RECORDS, 1990L);
+    writeAndValidate(schema, expectedRecords);
+  }
+
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedRecords) throws IOException {
+    RowType flinkSchema = FlinkSchemaUtil.convert(schema);
     List<RowData> expectedRows = Lists.newArrayList(RandomRowData.convert(schema, expectedRecords));
 
     File recordsFile = File.createTempFile("junit", null, temp.toFile());
@@ -104,4 +110,10 @@ protected void writeAndValidate(Schema schema) throws IOException {
       assertThat(records).isExhausted();
     }
   }
+
+  @Override
+  @Ignore("ORC file format supports null values even for required fields")
+  public void testWriteNullValueForRequiredType() {
+    super.testWriteNullValueForRequiredType();
+  }
 }

flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReader.java

@@ -236,4 +236,9 @@ protected void writeAndValidate(Schema schema) throws IOException {
         RandomGenericData.generateFallbackRecords(schema, NUM_RECORDS, 21124, NUM_RECORDS / 20),
         schema);
   }
+
+  @Override
+  protected void writeAndValidate(Schema schema, List<Record> expectedData) throws IOException {
+    writeAndValidate(expectedData, schema);
+  }
 }