This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 9fffeb63b1 Flink: Backport: Fix ConvertEqualityDeletes key
serialization for temporal equality fields (#18610) (#18720)
9fffeb63b1 is described below
commit 9fffeb63b1605a7c3acf218e58b7eeb43bdb1b01
Author: yhyyz <[email protected]>
AuthorDate: Mon Nov 2 17:32:40 2026 +0800
Flink: Backport: Fix ConvertEqualityDeletes key serialization for temporal
equality fields (#18610) (#18720)
---
.../operator/EqualityConvertReader.java | 11 +++-
.../maintenance/operator/StructLikeSerializer.java | 9 +++
.../maintenance/operator/OperatorTestBase.java | 10 +++-
.../operator/TestEqualityConvertReader.java | 64 ++++++++++++++++++++++
.../operator/EqualityConvertReader.java | 11 +++-
.../maintenance/operator/StructLikeSerializer.java | 9 +++
.../maintenance/operator/OperatorTestBase.java | 10 +++-
.../operator/TestEqualityConvertReader.java | 64 ++++++++++++++++++++++
8 files changed, 180 insertions(+), 8 deletions(-)
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
index a0c1752fdf..a46e100293 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
@@ -38,6 +38,7 @@ import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.BaseDeleteLoader;
import org.apache.iceberg.data.DeleteLoader;
+import org.apache.iceberg.data.InternalRecordWrapper;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.deletes.PositionDeleteIndex;
import org.apache.iceberg.flink.TableLoader;
@@ -75,6 +76,9 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
private transient Table table;
private transient Schema keySchema;
private transient StructLikeSerializer fieldSerializer;
+ // Generic records hold java.time values; the serializer expects Iceberg's
internal
+ // representation (e.g. days for DATE, micros for TIMESTAMP).
+ private transient InternalRecordWrapper keyWrapper;
private transient DeleteLoader deleteLoader;
public EqualityConvertReader(
@@ -102,6 +106,7 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
"Equality field IDs %s not present in table schema",
eqFieldIds);
fieldSerializer = new StructLikeSerializer();
+ keyWrapper = new InternalRecordWrapper(keySchema.asStruct());
deleteLoader = new BaseDeleteLoader(deleteFile ->
table.io().newInputFile(deleteFile));
}
@@ -165,7 +170,8 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
continue;
}
- SerializedEqualityValues key = fieldSerializer.serializeKey(record,
keySchema.asStruct());
+ SerializedEqualityValues key =
+ fieldSerializer.serializeKey(keyWrapper.wrap(record),
keySchema.asStruct());
out.collect(
IndexCommand.addDataRow(
mainSnapshotId,
@@ -201,7 +207,8 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
FormatModelRegistry.readBuilder(file.format(), Record.class, input);
try (CloseableIterable<Record> records =
builder.project(keySchema).reuseContainers().build()) {
for (Record record : records) {
- SerializedEqualityValues key = fieldSerializer.serializeKey(record,
keySchema.asStruct());
+ SerializedEqualityValues key =
+ fieldSerializer.serializeKey(keyWrapper.wrap(record),
keySchema.asStruct());
out.collect(
IndexCommand.resolveDelete(
mainSnapshotId, indexGeneration, key, dataSequenceNumber,
deleteSpecId));
diff --git
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
index e04a7c53b5..43a6a520b8 100644
---
a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
+++
b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
@@ -50,6 +50,15 @@ class StructLikeSerializer {
private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
private final DataOutputStream dos = new DataOutputStream(baos);
+ /**
+ * Serializes the equality values of {@code key} into a keyed-state key.
+ *
+ * <p>Field values must use Iceberg's internal representation, as expected
by {@link
+ * Conversions#toByteBuffer(Type, Object)}: for example days as {@code int}
for {@code date},
+ * microseconds as {@code long} for {@code time} and {@code timestamp}.
Generic {@link
+ * org.apache.iceberg.data.Record}s hold {@code java.time} values instead
and must be wrapped with
+ * {@link org.apache.iceberg.data.InternalRecordWrapper} first.
+ */
public SerializedEqualityValues serializeKey(StructLike key,
Types.StructType keyType) {
baos.reset();
try {
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
index 3557902813..51072fef10 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
@@ -177,12 +177,18 @@ public class OperatorTestBase {
}
protected static Table createTableWithDelete(int formatVersion) {
+ return createTableWithDelete(
+ formatVersion, SCHEMA_WITH_PRIMARY_KEY, PartitionSpec.unpartitioned());
+ }
+
+ protected static Table createTableWithDelete(
+ int formatVersion, Schema schema, PartitionSpec spec) {
return CATALOG_EXTENSION
.catalog()
.createTable(
TestFixtures.TABLE_IDENTIFIER,
- SCHEMA_WITH_PRIMARY_KEY,
- PartitionSpec.unpartitioned(),
+ schema,
+ spec,
null,
ImmutableMap.of(
"format-version", String.valueOf(formatVersion),
"write.upsert.enabled", "true"));
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
index 9c93b8122a..543a08dc94 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
@@ -25,6 +25,11 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
import java.util.List;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
@@ -38,6 +43,7 @@ import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.FileHelpers;
import org.apache.iceberg.data.GenericAppenderHelper;
@@ -45,6 +51,8 @@ import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.DateTimeUtil;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -220,6 +228,62 @@ class TestEqualityConvertReader extends OperatorTestBase {
}
}
+ @Test
+ void serializesTemporalEqualityFieldsFromDataAndDeleteFiles() throws
Exception {
+ Schema schema =
+ new Schema(
+ Lists.newArrayList(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "d", Types.DateType.get()),
+ Types.NestedField.required(3, "t", Types.TimeType.get()),
+ Types.NestedField.required(4, "ts",
Types.TimestampType.withoutZone()),
+ Types.NestedField.required(5, "tstz",
Types.TimestampType.withZone())),
+ Sets.newHashSet(1, 2, 3, 4, 5));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("d").build();
+ Table table = createTableWithDelete(3, schema, spec);
+
+ LocalDate date = LocalDate.of(2026, 10, 1);
+ GenericRecord record = GenericRecord.create(schema);
+ record.setField("id", 1);
+ record.setField("d", date);
+ record.setField("t", LocalTime.of(12, 30, 15));
+ record.setField("ts", LocalDateTime.of(2026, 10, 1, 12, 30, 15));
+ record.setField("tstz", OffsetDateTime.of(2026, 10, 1, 12, 30, 15, 0,
ZoneOffset.UTC));
+
+ PartitionData partition = new PartitionData(spec.partitionType());
+ partition.set(0, DateTimeUtil.daysFromDate(date));
+
+ DataFile dataFile =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
+ .writeFile(partition, Lists.newArrayList(record));
+ table.newAppend().appendFile(dataFile).commit();
+
+ File file = File.createTempFile("junit", null, tempDir.toFile());
+ assertThat(file.delete()).isTrue();
+ DeleteFile eqDelete =
+ FileHelpers.writeDeleteFile(
+ table, Files.localOutput(file), partition,
Lists.newArrayList(record), schema);
+ table.refresh();
+
+ List<Integer> fieldIds = Lists.newArrayList(1, 2, 3, 4, 5);
+
+ try (OneInputStreamOperatorTestHarness<ReadCommand, IndexCommand> harness =
+ createHarness(fieldIds)) {
+ harness.open();
+
+ harness.processElement(
+ ReadCommand.stagingDataFile(new FlinkAddedRowsScanTask(dataFile,
spec), 0L, 0L, 0L), 0);
+ harness.processElement(ReadCommand.eqDeleteFile(eqDelete, spec, 0L, 0L,
0L), 1);
+
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNullOrEmpty();
+ List<IndexCommand> output = harness.extractOutputValues();
+ assertThat(output).hasSize(2);
+
assertThat(output.get(0).type()).isEqualTo(IndexCommand.Type.ADD_STAGING_DATA_ROW);
+
assertThat(output.get(1).type()).isEqualTo(IndexCommand.Type.RESOLVE_DELETE);
+ assertThat(output.get(0).key()).isEqualTo(output.get(1).key());
+ }
+ }
+
@Test
void propagatesMainSnapshotId() throws Exception {
Table table = createTableWithDelete(3);
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
index a0c1752fdf..a46e100293 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
@@ -38,6 +38,7 @@ import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.BaseDeleteLoader;
import org.apache.iceberg.data.DeleteLoader;
+import org.apache.iceberg.data.InternalRecordWrapper;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.deletes.PositionDeleteIndex;
import org.apache.iceberg.flink.TableLoader;
@@ -75,6 +76,9 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
private transient Table table;
private transient Schema keySchema;
private transient StructLikeSerializer fieldSerializer;
+ // Generic records hold java.time values; the serializer expects Iceberg's
internal
+ // representation (e.g. days for DATE, micros for TIMESTAMP).
+ private transient InternalRecordWrapper keyWrapper;
private transient DeleteLoader deleteLoader;
public EqualityConvertReader(
@@ -102,6 +106,7 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
"Equality field IDs %s not present in table schema",
eqFieldIds);
fieldSerializer = new StructLikeSerializer();
+ keyWrapper = new InternalRecordWrapper(keySchema.asStruct());
deleteLoader = new BaseDeleteLoader(deleteFile ->
table.io().newInputFile(deleteFile));
}
@@ -165,7 +170,8 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
continue;
}
- SerializedEqualityValues key = fieldSerializer.serializeKey(record,
keySchema.asStruct());
+ SerializedEqualityValues key =
+ fieldSerializer.serializeKey(keyWrapper.wrap(record),
keySchema.asStruct());
out.collect(
IndexCommand.addDataRow(
mainSnapshotId,
@@ -201,7 +207,8 @@ public class EqualityConvertReader extends
ProcessFunction<ReadCommand, IndexCom
FormatModelRegistry.readBuilder(file.format(), Record.class, input);
try (CloseableIterable<Record> records =
builder.project(keySchema).reuseContainers().build()) {
for (Record record : records) {
- SerializedEqualityValues key = fieldSerializer.serializeKey(record,
keySchema.asStruct());
+ SerializedEqualityValues key =
+ fieldSerializer.serializeKey(keyWrapper.wrap(record),
keySchema.asStruct());
out.collect(
IndexCommand.resolveDelete(
mainSnapshotId, indexGeneration, key, dataSequenceNumber,
deleteSpecId));
diff --git
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
index e04a7c53b5..43a6a520b8 100644
---
a/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
+++
b/flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
@@ -50,6 +50,15 @@ class StructLikeSerializer {
private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
private final DataOutputStream dos = new DataOutputStream(baos);
+ /**
+ * Serializes the equality values of {@code key} into a keyed-state key.
+ *
+ * <p>Field values must use Iceberg's internal representation, as expected
by {@link
+ * Conversions#toByteBuffer(Type, Object)}: for example days as {@code int}
for {@code date},
+ * microseconds as {@code long} for {@code time} and {@code timestamp}.
Generic {@link
+ * org.apache.iceberg.data.Record}s hold {@code java.time} values instead
and must be wrapped with
+ * {@link org.apache.iceberg.data.InternalRecordWrapper} first.
+ */
public SerializedEqualityValues serializeKey(StructLike key,
Types.StructType keyType) {
baos.reset();
try {
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
index 7c811b4d2e..287abe7cc0 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/OperatorTestBase.java
@@ -177,12 +177,18 @@ public class OperatorTestBase {
}
protected static Table createTableWithDelete(int formatVersion) {
+ return createTableWithDelete(
+ formatVersion, SCHEMA_WITH_PRIMARY_KEY, PartitionSpec.unpartitioned());
+ }
+
+ protected static Table createTableWithDelete(
+ int formatVersion, Schema schema, PartitionSpec spec) {
return CATALOG_EXTENSION
.catalog()
.createTable(
TestFixtures.TABLE_IDENTIFIER,
- SCHEMA_WITH_PRIMARY_KEY,
- PartitionSpec.unpartitioned(),
+ schema,
+ spec,
null,
ImmutableMap.of(
"format-version", String.valueOf(formatVersion),
"write.upsert.enabled", "true"));
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
index 9c93b8122a..543a08dc94 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertReader.java
@@ -25,6 +25,11 @@ import static
org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
import java.nio.file.Path;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
import java.util.List;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.ProcessFunctionTestHarnesses;
@@ -38,6 +43,7 @@ import org.apache.iceberg.ManifestFiles;
import org.apache.iceberg.ManifestReader;
import org.apache.iceberg.PartitionData;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.data.FileHelpers;
import org.apache.iceberg.data.GenericAppenderHelper;
@@ -45,6 +51,8 @@ import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.DateTimeUtil;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -220,6 +228,62 @@ class TestEqualityConvertReader extends OperatorTestBase {
}
}
+ @Test
+ void serializesTemporalEqualityFieldsFromDataAndDeleteFiles() throws
Exception {
+ Schema schema =
+ new Schema(
+ Lists.newArrayList(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "d", Types.DateType.get()),
+ Types.NestedField.required(3, "t", Types.TimeType.get()),
+ Types.NestedField.required(4, "ts",
Types.TimestampType.withoutZone()),
+ Types.NestedField.required(5, "tstz",
Types.TimestampType.withZone())),
+ Sets.newHashSet(1, 2, 3, 4, 5));
+ PartitionSpec spec =
PartitionSpec.builderFor(schema).identity("d").build();
+ Table table = createTableWithDelete(3, schema, spec);
+
+ LocalDate date = LocalDate.of(2026, 10, 1);
+ GenericRecord record = GenericRecord.create(schema);
+ record.setField("id", 1);
+ record.setField("d", date);
+ record.setField("t", LocalTime.of(12, 30, 15));
+ record.setField("ts", LocalDateTime.of(2026, 10, 1, 12, 30, 15));
+ record.setField("tstz", OffsetDateTime.of(2026, 10, 1, 12, 30, 15, 0,
ZoneOffset.UTC));
+
+ PartitionData partition = new PartitionData(spec.partitionType());
+ partition.set(0, DateTimeUtil.daysFromDate(date));
+
+ DataFile dataFile =
+ new GenericAppenderHelper(table, FileFormat.PARQUET, tempDir)
+ .writeFile(partition, Lists.newArrayList(record));
+ table.newAppend().appendFile(dataFile).commit();
+
+ File file = File.createTempFile("junit", null, tempDir.toFile());
+ assertThat(file.delete()).isTrue();
+ DeleteFile eqDelete =
+ FileHelpers.writeDeleteFile(
+ table, Files.localOutput(file), partition,
Lists.newArrayList(record), schema);
+ table.refresh();
+
+ List<Integer> fieldIds = Lists.newArrayList(1, 2, 3, 4, 5);
+
+ try (OneInputStreamOperatorTestHarness<ReadCommand, IndexCommand> harness =
+ createHarness(fieldIds)) {
+ harness.open();
+
+ harness.processElement(
+ ReadCommand.stagingDataFile(new FlinkAddedRowsScanTask(dataFile,
spec), 0L, 0L, 0L), 0);
+ harness.processElement(ReadCommand.eqDeleteFile(eqDelete, spec, 0L, 0L,
0L), 1);
+
+
assertThat(harness.getSideOutput(TaskResultAggregator.ERROR_STREAM)).isNullOrEmpty();
+ List<IndexCommand> output = harness.extractOutputValues();
+ assertThat(output).hasSize(2);
+
assertThat(output.get(0).type()).isEqualTo(IndexCommand.Type.ADD_STAGING_DATA_ROW);
+
assertThat(output.get(1).type()).isEqualTo(IndexCommand.Type.RESOLVE_DELETE);
+ assertThat(output.get(0).key()).isEqualTo(output.get(1).key());
+ }
+ }
+
@Test
void propagatesMainSnapshotId() throws Exception {
Table table = createTableWithDelete(3);
