Fix the broken TestFlinkUpsert.

Author: openinx

flink/v1.14/flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriterFactory.java

@@ -90,6 +90,7 @@ public RowDataTaskWriterFactory(
     if (equalityFieldIds == null || equalityFieldIds.isEmpty()) {
       this.deleteSchema = null;
       this.fileWriterFactory = FlinkFileWriterFactory.builderFor(table)
+          .dataSchema(schema)
           .dataFlinkType(flinkSchema)
           .build();
     } else if (upsert) {
@@ -98,13 +99,15 @@ public RowDataTaskWriterFactory(
       // that are correct for the deleted row. Therefore, only write the equality delete fields.
       this.deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
       this.fileWriterFactory = FlinkFileWriterFactory.builderFor(table)
+          .dataSchema(schema)
           .dataFlinkType(flinkSchema)
           .equalityFieldIds(ArrayUtil.toIntArray(equalityFieldIds))
           .equalityDeleteRowSchema(deleteSchema)
           .build();
     } else {
       this.deleteSchema = TypeUtil.select(schema, Sets.newHashSet(equalityFieldIds));
       this.fileWriterFactory = FlinkFileWriterFactory.builderFor(table)
+          .dataSchema(schema)
           .dataFlinkType(flinkSchema)
           .equalityFieldIds(ArrayUtil.toIntArray(equalityFieldIds))
           .equalityDeleteRowSchema(schema)