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

mxm · 2025-01-22T16:40:10Z

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.

mxm · 2025-01-22T16:41:16Z

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

+      // This will produce incorrect writes instead of failing with a NullPointerException.
+      if (struct.isNullAt(index)) {
+        return null;
+      }


Actual fix is here.

do we need to fix the FlinkOrcWriters?

You are right. I revised the approach because it was too easy to miss instances. I'm instead wrapping RowData#createFieldGetter to make sure to null-check also for required / NonNull types. I'll raise an issue on the Flink side as well.

Also see #12049 (comment).

pvary · 2025-01-23T06:29:27Z

@mxm: Please remove the 1.18, 1.19 changes from the PR. It is much easier to review this way, and apply changes required by the reviewer. When the PR has been merged, we can backport the changes to the other Flink versions.

QQ: What happens when we have a type discrepancy between the Iceberg type and the RawData type? Could we have issues with other conversions? Do we have a way to prevent those?

mxm · 2025-01-23T11:24:45Z

@mxm: Please remove the 1.18, 1.19 changes from the PR. It is much easier to review this way, and apply changes required by the reviewer. When the PR has been merged, we can backport the changes to the other Flink versions.

Makes sense! Done.

QQ: What happens when we have a type discrepancy between the Iceberg type and the RawData type? Could we have issues with other conversions? Do we have a way to prevent those?

Type discrepancies between Iceberg and Flink types will error in Flink's TypeSerializer for a given field. For example, an int field will use IntSerializer which only accepts Integer. This will raise an NoSuchMethodError during serialization. As long as we use the same serializer also for deserialization, we should be fine. That is the case.

mxm · 2025-01-23T11:28:01Z

@pvary I had to re-add the 1.18 and 1.19 changes, but they are in a separate commit. The reason is that I modified a test base class which affects also 1.18 and 1.19. We can't build otherwise.

mxm · 2025-01-23T15:59:56Z

Tests are green.

mxm · 2025-01-23T17:36:40Z

CC @stevenzwu

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

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

mxm · 2025-01-27T12:59:22Z

Rebased.

mxm · 2025-01-27T16:03:55Z

Flaky Spark test, otherwise passing.

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

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

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

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

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

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

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergSink.java

mxm · 2025-01-31T20:43:44Z

flink/v1.18/flink/src/main/java/org/apache/iceberg/flink/FlinkRowData.java

+  public static RowData.FieldGetter createFieldGetter(LogicalType fieldType, int fieldPos) {
+    RowData.FieldGetter flinkFieldGetter = RowData.createFieldGetter(fieldType, fieldPos);
+    return rowData -> {
+      // Be sure to check for null values, even if the field is required. Flink
+      // RowData.createFieldGetter(..) does not null-check optional / nullable types. Without this
+      // explicit null check, the null flag of BinaryRowData will be ignored and random bytes will
+      // be parsed as actual values. This will produce incorrect writes instead of failing with a
+      // NullPointerException.
+      if (!fieldType.isNullable() && rowData.isNullAt(fieldPos)) {
+        return null;
+      }
+      return flinkFieldGetter.getFieldOrNull(rowData);
+    };
+  }


Moved the core change here to replace all the RowData#createFieldGetter calls. The idea is to also perform a null check for non-null types to prevent interpreting nulled fields in BinaryRowData as actual values. Unfortunately, Flink itself only adds the null check for nullable types and defers additional null checks to the caller. I'll report this in upstream Flink as well.

I like this new approach.

I am also wondering if this is also a bug in Flink where RowData.createFieldGetter couldn't handle the BinaryRowData properly regarding null value.

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.

stevenzwu · 2025-01-31T23:15:21Z

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

+
+  @Test
+  public void testWriteNullValueForRequiredType() {
+    Assumptions.assumeThat(supportsDefaultValues()).isTrue();


where is supportsDefaultValues() defined?

github-actions bot added data flink labels Jan 22, 2025

mxm commented Jan 22, 2025

View reviewed changes

mxm force-pushed the null-value-check branch from 5ecff96 to f763c54 Compare January 22, 2025 18:01

mxm force-pushed the null-value-check branch 3 times, most recently from b8700c9 to 734b7bb Compare January 23, 2025 11:18

mxm force-pushed the null-value-check branch 4 times, most recently from f4893cc to 59dacfb Compare January 23, 2025 13:58

pvary reviewed Jan 24, 2025

View reviewed changes

data/src/test/java/org/apache/iceberg/data/orc/TestGenericData.java Outdated Show resolved Hide resolved

pvary reviewed Jan 24, 2025

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java Outdated Show resolved Hide resolved

mxm force-pushed the null-value-check branch from e743c35 to 133b047 Compare January 27, 2025 12:57

mxm force-pushed the null-value-check branch 3 times, most recently from def98b6 to 46f8c23 Compare January 27, 2025 13:47

mxm force-pushed the null-value-check branch from 46f8c23 to d3d4e46 Compare January 28, 2025 09:52

pvary reviewed Jan 28, 2025

View reviewed changes

data/src/test/java/org/apache/iceberg/data/DataTest.java Outdated Show resolved Hide resolved

mxm force-pushed the null-value-check branch 2 times, most recently from bbc627a to 8f90712 Compare January 28, 2025 17:38

stevenzwu reviewed Jan 28, 2025

View reviewed changes

mxm force-pushed the null-value-check branch from 8f90712 to 0e96bea Compare January 31, 2025 19:48

mxm force-pushed the null-value-check branch 6 times, most recently from 6b69d52 to ead98ae Compare January 31, 2025 20:42

mxm commented Jan 31, 2025

View reviewed changes

mxm added 2 commits January 31, 2025 22:20

Re-add 1.18 and 1.19 changes because they inherit base class changes

c4aba0b

mxm force-pushed the null-value-check branch from ead98ae to c4aba0b Compare January 31, 2025 21:20

stevenzwu reviewed Jan 31, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

mxm commented Jan 22, 2025

mxm Jan 22, 2025

stevenzwu Jan 28, 2025

mxm Jan 31, 2025

mxm Jan 31, 2025

pvary commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 27, 2025

mxm commented Jan 27, 2025

mxm Jan 31, 2025

stevenzwu Jan 31, 2025

stevenzwu Jan 31, 2025

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

Are you sure you want to change the base?

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

Conversation

mxm commented Jan 22, 2025

mxm Jan 22, 2025

Choose a reason for hiding this comment

stevenzwu Jan 28, 2025

Choose a reason for hiding this comment

mxm Jan 31, 2025

Choose a reason for hiding this comment

mxm Jan 31, 2025

Choose a reason for hiding this comment

pvary commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 23, 2025

mxm commented Jan 27, 2025

mxm commented Jan 27, 2025

mxm Jan 31, 2025

Choose a reason for hiding this comment

stevenzwu Jan 31, 2025

Choose a reason for hiding this comment

stevenzwu Jan 31, 2025

Choose a reason for hiding this comment