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

Flink's BinaryRowData uses a magic byte to indicate null values in the backing
byte arrays. Flink's internal RowData#createFieldGetter method which Iceberg
uses, only adds a null check whenever a type is nullable. We map Iceberg's
optional attribute to nullable, but Iceberg's required attribute to
non-nullable. The latter creates an issue when the user, by mistake, nulls a
field. The resulting RowData field will then be interpreted as actual data
because the null field is not checked. This yields random values which should
have been null and produced an error in the writer.

The solution is to always check if a field is nullable before attempting to read
data from it.

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

@@ -20,6 +20,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.google.common.collect.Lists;
 import java.io.File;
 import java.io.IOException;
 import java.util.List;
@@ -32,17 +33,28 @@
 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.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

@@ -22,6 +22,7 @@
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.google.common.collect.Lists;
 import java.io.File;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
@@ -42,14 +43,14 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.orc.ORC;
-import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.OrcFile;
 import org.apache.orc.TypeDescription;
 import org.apache.orc.Writer;
 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.20/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.20/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.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java

@@ -20,6 +20,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.google.common.collect.Lists;
 import java.io.File;
 import java.io.IOException;
 import java.util.Iterator;
@@ -38,15 +39,20 @@
 import org.apache.iceberg.io.CloseableIterable;
 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.20/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.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetWriter.java

@@ -20,11 +20,15 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.google.common.collect.Lists;
 import java.io.File;
 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,6 +37,7 @@
 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;
@@ -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);
+  }
 }