You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/07/07 09:08:58 UTC
(iceberg) branch main updated: Flink: Resolve unpartitioned equality deletes across all partitions (#17018)
This is an automated email from the ASF dual-hosted git repository.
pvary 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 34b03b7710 Flink: Resolve unpartitioned equality deletes across all partitions (#17018)
34b03b7710 is described below
commit 34b03b7710a116a7ffc14b2e68aad1b8ee484d1f
Author: Maximilian Michels <mx...@apache.org>
AuthorDate: Tue Jul 7 11:08:49 2026 +0200
Flink: Resolve unpartitioned equality deletes across all partitions (#17018)
---
.../operator/EqualityConvertPKIndex.java | 28 +++++-
.../operator/EqualityConvertReader.java | 16 +--
.../flink/maintenance/operator/IndexCommand.java | 24 ++++-
.../maintenance/operator/StructLikeSerializer.java | 11 +--
.../api/TestConvertEqualityDeletes.java | 71 ++++++++++++++
.../operator/TestEqualityConvertPKIndex.java | 108 +++++++++++++++++----
.../operator/TestStructLikeSerializer.java | 36 +++----
7 files changed, 233 insertions(+), 61 deletions(-)
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
index e688142add..5c949122d9 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertPKIndex.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.flink.maintenance.operator;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.state.ListState;
@@ -34,6 +35,7 @@ import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.util.Collector;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -97,6 +99,8 @@ public class EqualityConvertPKIndex
new ValueStateDescriptor<>("resolveTimestamp", Types.LONG);
private static final ValueStateDescriptor<Long> RESOLVE_SEQUENCE_NUMBER_DESCRIPTOR =
new ValueStateDescriptor<>("resolveSequenceNumber", Types.LONG);
+ private static final ListStateDescriptor<Integer> RESOLVE_SPEC_IDS_DESCRIPTOR =
+ new ListStateDescriptor<>("resolveSpecIds", Types.INT);
private transient ValueState<Long> mainSequenceVersion;
// Resolvable rows for this key. Populated immediately for main data, or from onTimer for
@@ -111,6 +115,9 @@ public class EqualityConvertPKIndex
// Maximum sequence number among the cycle's RESOLVE_DELETEs for this key. Deletes a
// data row only when the row's sequence number is below it.
private transient ValueState<Long> resolveSequenceNumber;
+ // Spec ids of the cycle's RESOLVE_DELETEs for this key. A delete phase may carry same-key deletes
+ // from multiple specs, and each scope must apply.
+ private transient ListState<Integer> resolveSpecIds;
private transient Counter resolvedDeleteNumCounter;
private transient Counter indexedKeyNumCounter;
@@ -130,6 +137,7 @@ public class EqualityConvertPKIndex
bufferedRows = getRuntimeContext().getListState(BUFFERED_ROWS_DESCRIPTOR);
resolveTimestamp = getRuntimeContext().getState(RESOLVE_TIMESTAMP_DESCRIPTOR);
resolveSequenceNumber = getRuntimeContext().getState(RESOLVE_SEQUENCE_NUMBER_DESCRIPTOR);
+ resolveSpecIds = getRuntimeContext().getListState(RESOLVE_SPEC_IDS_DESCRIPTOR);
resolvedDeleteNumCounter =
getRuntimeContext().getMetricGroup().counter(RESOLVED_DELETE_NUM_METRIC);
indexedKeyNumCounter = getRuntimeContext().getMetricGroup().counter(INDEXED_KEY_NUM_METRIC);
@@ -174,6 +182,10 @@ public class EqualityConvertPKIndex
resolveSequenceNumber.update(cmd.deleteSequenceNumber());
}
+ // Accumulate every delete's scope.
+ // One delete phase can carry same-key deletes from multiple specs.
+ resolveSpecIds.add(cmd.deleteSpecId());
+
ctx.timerService().registerEventTimeTimer(ts);
} else {
throw new IllegalStateException("Unexpected command type in keyed stream: " + cmd.type());
@@ -223,6 +235,7 @@ public class EqualityConvertPKIndex
resolveDeletes(out);
resolveTimestamp.clear();
resolveSequenceNumber.clear();
+ resolveSpecIds.clear();
} else {
applyBufferedRows();
}
@@ -250,14 +263,22 @@ public class EqualityConvertPKIndex
// indexed sequence is not comparable to the staging delete's; event-time ordering already
// prevents false deletion.
Long deleteSeq = resolveSequenceNumber.value();
+ // A partitioned delete applies only to data rows of the same spec. An unpartitioned delete
+ // (GLOBAL_DELETE_SPEC_ID) applies to every spec. deleteSpecIds holds every scope seen this
+ // cycle (usually one).
+ Set<Integer> deleteSpecIds = Sets.newHashSet(resolveSpecIds.get());
+ boolean globalDelete = deleteSpecIds.contains(IndexCommand.GLOBAL_DELETE_SPEC_ID);
List<DVPosition> nonEligible = Lists.newArrayList();
int count = 0;
for (DVPosition pos : dataRowPositions.get()) {
- if (stagingOnTargetBranch && deleteSeq != null && pos.dataSequenceNumber() >= deleteSeq) {
- nonEligible.add(pos);
- } else {
+ boolean specMatch = globalDelete || deleteSpecIds.contains(pos.specId());
+ boolean sequenceMatch =
+ !stagingOnTargetBranch || deleteSeq == null || pos.dataSequenceNumber() < deleteSeq;
+ if (specMatch && sequenceMatch) {
out.collect(pos);
count++;
+ } else {
+ nonEligible.add(pos);
}
}
@@ -270,5 +291,6 @@ public class EqualityConvertPKIndex
bufferedRows.clear();
resolveTimestamp.clear();
resolveSequenceNumber.clear();
+ resolveSpecIds.clear();
}
}
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
index ccc47a015c..809e60a8f8 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/EqualityConvertReader.java
@@ -151,7 +151,9 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
int specId = file.specId();
StructLike partition = file.partition();
- Types.StructType partitionType = table.specs().get(specId).partitionType();
+ // Use the planner-serialized task spec, not the reader's table, which is loaded once and may
+ // lack specs added after startup.
+ Types.StructType partitionType = task.spec().partitionType();
byte[] partitionBytes = fieldSerializer.encodePartition(partition, partitionType);
InputFile input = table.io().newInputFile(file.location());
@@ -167,8 +169,7 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
continue;
}
- SerializedEqualityValues key =
- fieldSerializer.serializeKey(record, keySchema.asStruct(), specId);
+ SerializedEqualityValues key = fieldSerializer.serializeKey(record, keySchema.asStruct());
out.collect(
IndexCommand.addDataRow(
mainSnapshotId,
@@ -194,17 +195,20 @@ public class EqualityConvertReader extends ProcessFunction<ReadCommand, IndexCom
ContentFile<?> file = task.file();
int specId = file.specId();
+ // An unpartitioned equality delete applies globally; a partitioned one applies only within its
+ // own spec. Use the planner-serialized task spec, not the reader's table, which is loaded once
+ // and may lack specs added after startup.
+ int deleteSpecId = task.spec().isUnpartitioned() ? IndexCommand.GLOBAL_DELETE_SPEC_ID : specId;
InputFile input = table.io().newInputFile(file.location());
ReadBuilder<Record, Schema> builder =
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(), specId);
+ SerializedEqualityValues key = fieldSerializer.serializeKey(record, keySchema.asStruct());
out.collect(
IndexCommand.resolveDelete(
- mainSnapshotId, mainSequenceNumber, key, dataSequenceNumber));
+ mainSnapshotId, mainSequenceNumber, key, dataSequenceNumber, deleteSpecId));
}
}
}
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
index 160d563607..509812cbb3 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/IndexCommand.java
@@ -40,6 +40,9 @@ import org.apache.flink.annotation.Internal;
* <p>{@link #rowPosition} is the data row's location, set for the two add types and null otherwise;
* the data sequence number it carries lets the worker apply a delete only to older rows. {@code
* deleteSequenceNumber} is the equality delete's sequence number, set for {@link
+ * Type#RESOLVE_DELETE} and unused (-1) otherwise. {@code deleteSpecId} is the partition spec id of
+ * the delete file, used to scope a partitioned delete to data rows of the same spec; {@link
+ * #GLOBAL_DELETE_SPEC_ID} marks an unpartitioned delete that applies to every spec. Set for {@link
* Type#RESOLVE_DELETE} and unused (-1) otherwise.
*/
@Internal
@@ -49,9 +52,13 @@ public record IndexCommand(
Long mainSequenceNumber,
SerializedEqualityValues key,
DVPosition rowPosition,
- long deleteSequenceNumber)
+ long deleteSequenceNumber,
+ int deleteSpecId)
implements Serializable {
+ /** Spec id sentinel for an unpartitioned equality delete, which applies as a global delete. */
+ public static final int GLOBAL_DELETE_SPEC_ID = -1;
+
public enum Type {
ADD_DATA_ROW,
ADD_STAGING_DATA_ROW,
@@ -75,6 +82,7 @@ public record IndexCommand(
mainSequenceNumber,
key,
new DVPosition(filePath, position, specId, partition, dataSequenceNumber),
+ -1,
-1);
}
@@ -82,12 +90,20 @@ public record IndexCommand(
Long mainSnapshotId,
Long mainSequenceNumber,
SerializedEqualityValues key,
- long deleteSequenceNumber) {
+ long deleteSequenceNumber,
+ int deleteSpecId) {
return new IndexCommand(
- Type.RESOLVE_DELETE, mainSnapshotId, mainSequenceNumber, key, null, deleteSequenceNumber);
+ Type.RESOLVE_DELETE,
+ mainSnapshotId,
+ mainSequenceNumber,
+ key,
+ null,
+ deleteSequenceNumber,
+ deleteSpecId);
}
public static IndexCommand clearBeforeReindex(long mainSnapshotId, long mainSequenceNumber) {
- return new IndexCommand(Type.CLEAR_INDEX, mainSnapshotId, mainSequenceNumber, null, null, -1);
+ return new IndexCommand(
+ Type.CLEAR_INDEX, mainSnapshotId, mainSequenceNumber, null, null, -1, -1);
}
}
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
index 83024d9795..e04a7c53b5 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/maintenance/operator/StructLikeSerializer.java
@@ -36,8 +36,10 @@ import org.apache.iceberg.types.Types;
* Serializer for {@link StructLike} tuples. Two entry points:
*
* <ul>
- * <li>{@link #serializeKey} builds a Flink keyed-state key ({@link SerializedEqualityValues}),
- * prefixed with the partition spec id so partition evolution keeps specs disjoint.
+ * <li>{@link #serializeKey} builds a Flink keyed-state key ({@link SerializedEqualityValues})
+ * from the equality values alone, so all rows with the same key co-locate on one shard
+ * regardless of partition spec. Spec scoping of a delete is applied at resolve time, not in
+ * the key.
* <li>{@link #encodePartition} / {@link #decodePartition} serialize partition tuples into bytes.
* </ul>
*/
@@ -48,12 +50,9 @@ class StructLikeSerializer {
private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
private final DataOutputStream dos = new DataOutputStream(baos);
- public SerializedEqualityValues serializeKey(
- StructLike key, Types.StructType keyType, int specId) {
+ public SerializedEqualityValues serializeKey(StructLike key, Types.StructType keyType) {
baos.reset();
try {
- dos.writeInt(specId);
-
List<Types.NestedField> fields = keyType.fields();
dos.writeInt(fields.size());
for (Types.NestedField field : fields) {
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
index 1c2f230315..d1ab299544 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/api/TestConvertEqualityDeletes.java
@@ -734,6 +734,62 @@ class TestConvertEqualityDeletes extends MaintenanceTaskTestBase {
assertRecords(table, ImmutableList.of(createRecord(1, "b"), createRecord(2, "a")));
}
+ @Test
+ void testUnpartitionedDeleteAppliesAcrossPartitionEvolution() throws Exception {
+ // Data is written under a partitioned spec, then the table evolves to unpartitioned. A delete
+ // written under the unpartitioned spec is a global delete and must remove the data row from the
+ // older spec. This only works because the index key is spec-independent (spec scoping happens
+ // at resolve time, where a global delete matches every spec).
+ Table table = createPartitionedTableWithDelete(3);
+ insertPartitioned(table, 1, "a");
+
+ table.updateSpec().removeField("data").commit();
+ table.refresh();
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ // writeEqualityDelete writes with the current (now unpartitioned) spec, so the delete is
+ // global.
+ DeleteFile eqDelete = writeEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ appendConvertTask();
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ table.refresh();
+ assertRecords(table, ImmutableList.of());
+ }
+
+ @Test
+ void testPartitionedDeleteDoesNotApplyAcrossPartitionEvolution() throws Exception {
+ // Inverse of the global case: a partitioned delete under a new spec must NOT remove a data row
+ // written under an older spec. The row survives because resolution is scoped to the delete's
+ // spec.
+ Table table = createPartitionedTableWithDelete(3);
+ insertPartitioned(table, 1, "a");
+
+ // Evolve to a different partition spec so the delete below is written under a new spec id.
+ table.updateSpec().removeField("data").addField("id").commit();
+ table.refresh();
+
+ table.manageSnapshots().createBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ DeleteFile eqDelete = writeIdPartitionedEqualityDelete(table, 1, "a");
+ table.newRowDelta().addDeletes(eqDelete).toBranch(STAGING_BRANCH).commit();
+ table.refresh();
+
+ appendConvertTask();
+ runAndWaitForSuccess(infra.env(), infra.source(), infra.sink());
+
+ table.refresh();
+ // The spec-0 row survives: the delete's spec differs and it is not a global (unpartitioned)
+ // delete.
+ assertRecords(table, ImmutableList.of(createRecord(1, "a")));
+ }
+
@Test
void testStagingPositionDeleteMergedIntoConversionDV() throws Exception {
Table table = createTableWithDelete(3);
@@ -1001,6 +1057,21 @@ class TestConvertEqualityDeletes extends MaintenanceTaskTestBase {
table.schema());
}
+ private DeleteFile writeIdPartitionedEqualityDelete(Table table, Integer id, String data)
+ throws IOException {
+ File file = File.createTempFile("junit", null, tempDir.toFile());
+ assertThat(file.delete()).isTrue();
+
+ PartitionData partition = new PartitionData(table.spec().partitionType());
+ partition.set(0, id);
+ return FileHelpers.writeDeleteFile(
+ table,
+ Files.localOutput(file),
+ partition,
+ Lists.newArrayList(createRecord(id, data)),
+ table.schema());
+ }
+
private static long dataFileCount(Table table) {
table.refresh();
long count = 0;
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPKIndex.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPKIndex.java
index 90b338458f..3af401f3c8 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPKIndex.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestEqualityConvertPKIndex.java
@@ -22,11 +22,15 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Collections;
import java.util.List;
+import java.util.stream.Stream;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.operators.co.CoBroadcastWithKeyedOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedBroadcastOperatorTestHarness;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
class TestEqualityConvertPKIndex {
@@ -42,6 +46,8 @@ class TestEqualityConvertPKIndex {
// Data-row sequence below the delete sequence, so existing tests still tombstone on resolve.
private static final long DATA_SEQ = 1L;
private static final long DELETE_SEQ = 2L;
+ // Default delete scope for the single-spec tests: global, so it resolves rows of any spec.
+ private static final int GLOBAL = IndexCommand.GLOBAL_DELETE_SPEC_ID;
@Test
void addDataRowProducesNoOutput() throws Exception {
@@ -90,7 +96,7 @@ class TestEqualityConvertPKIndex {
0));
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(1);
@@ -136,7 +142,7 @@ class TestEqualityConvertPKIndex {
0));
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(1);
@@ -156,6 +162,61 @@ class TestEqualityConvertPKIndex {
}
}
+ @ParameterizedTest(name = "deleteSpecs={0} resolvedSpecs={1}")
+ @MethodSource("deleteSpecScopes")
+ void resolvesDeletesScopedBySpec(List<Integer> deleteSpecIds, List<Integer> resolvedSpecIds)
+ throws Exception {
+ try (KeyedBroadcastOperatorTestHarness<
+ SerializedEqualityValues, IndexCommand, IndexCommand, DVPosition>
+ harness = createHarness()) {
+ harness.open();
+
+ // Same key, one row under each of two specs. Partition evolution co-locates them on one
+ // shard.
+ harness.processElement(new StreamRecord<>(addRow("spec0.parquet", 0, 0), 0));
+ harness.processElement(new StreamRecord<>(addRow("spec1.parquet", 1, 1), 0));
+ // All deletes fall in one delete phase (ts 1), so their scopes accumulate.
+ for (int deleteSpecId : deleteSpecIds) {
+ harness.processElement(
+ new StreamRecord<>(
+ IndexCommand.resolveDelete(
+ MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, deleteSpecId),
+ 1));
+ }
+
+ harness.watermark(1);
+
+ assertThat(harness.extractOutputValues())
+ .extracting(DVPosition::specId)
+ .containsExactlyInAnyOrderElementsOf(resolvedSpecIds);
+ }
+ }
+
+ private static Stream<Arguments> deleteSpecScopes() {
+ return Stream.of(
+ // A partitioned delete resolves only same-spec rows.
+ Arguments.of(List.of(1), List.of(1)),
+ // An unpartitioned (global) delete resolves rows of every spec.
+ Arguments.of(List.of(GLOBAL), List.of(0, 1)),
+ // Two partitioned deletes from different specs in one phase both apply.
+ Arguments.of(List.of(0, 1), List.of(0, 1)),
+ // A global delete scope affects a later partitioned delete in the same phase.
+ Arguments.of(List.of(GLOBAL, 0), List.of(0, 1)));
+ }
+
+ private static IndexCommand addRow(String filePath, long position, int specId) {
+ return IndexCommand.addDataRow(
+ MAIN_SNAPSHOT,
+ MAIN_SEQ,
+ KEY_1,
+ filePath,
+ position,
+ specId,
+ EMPTY_PARTITION,
+ DATA_SEQ,
+ false);
+ }
+
@Test
void resolveDeleteClearsEntries() throws Exception {
try (KeyedBroadcastOperatorTestHarness<
@@ -178,13 +239,13 @@ class TestEqualityConvertPKIndex {
0));
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(1);
assertThat(harness.extractOutputValues()).hasSize(1);
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 2));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 2));
harness.watermark(2);
assertThat(harness.extractOutputValues()).hasSize(1);
}
@@ -199,7 +260,7 @@ class TestEqualityConvertPKIndex {
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(1);
assertThat(harness.extractOutputValues()).isEmpty();
@@ -229,7 +290,9 @@ class TestEqualityConvertPKIndex {
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(
+ NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL),
+ 1));
harness.watermark(1);
assertThat(harness.extractOutputValues()).isEmpty();
@@ -266,7 +329,9 @@ class TestEqualityConvertPKIndex {
// Resolving KEY_1 against the (now-empty) index emits nothing.
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ), 2));
+ IndexCommand.resolveDelete(
+ NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL),
+ 2));
harness.watermark(2);
assertThat(harness.extractOutputValues()).isEmpty();
@@ -299,7 +364,9 @@ class TestEqualityConvertPKIndex {
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ), 2));
+ IndexCommand.resolveDelete(
+ NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL),
+ 2));
harness.watermark(2);
List<DVPosition> output = harness.extractOutputValues();
@@ -350,7 +417,9 @@ class TestEqualityConvertPKIndex {
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ), 3));
+ IndexCommand.resolveDelete(
+ NEW_MAIN_SNAPSHOT, NEW_MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL),
+ 3));
harness.watermark(3);
List<DVPosition> output = harness.extractOutputValues();
@@ -383,7 +452,7 @@ class TestEqualityConvertPKIndex {
new StreamRecord<>(IndexCommand.clearBeforeReindex(999L, 10L), 1));
harness.processElement(
- new StreamRecord<>(IndexCommand.resolveDelete(50L, 20L, KEY_1, DELETE_SEQ), 2));
+ new StreamRecord<>(IndexCommand.resolveDelete(50L, 20L, KEY_1, DELETE_SEQ, GLOBAL), 2));
harness.watermark(2);
List<DVPosition> output = harness.extractOutputValues();
@@ -429,7 +498,7 @@ class TestEqualityConvertPKIndex {
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(1);
List<DVPosition> output = harness.extractOutputValues();
@@ -463,7 +532,7 @@ class TestEqualityConvertPKIndex {
0));
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.processElement(
new StreamRecord<>(
IndexCommand.addDataRow(
@@ -487,7 +556,7 @@ class TestEqualityConvertPKIndex {
// A later cycle's delete (ts 3) removes the now-indexed new row.
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 3));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 3));
harness.watermark(3);
assertThat(harness.extractOutputValues()).hasSize(2);
assertThat(harness.extractOutputValues().get(1).dataFilePath()).isEqualTo("new.parquet");
@@ -518,7 +587,7 @@ class TestEqualityConvertPKIndex {
2));
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(2);
assertThat(harness.extractOutputValues()).isEmpty();
@@ -560,7 +629,7 @@ class TestEqualityConvertPKIndex {
0));
harness.processElement(
new StreamRecord<>(
- IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ), 1));
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, DELETE_SEQ, GLOBAL), 1));
harness.watermark(1);
@@ -601,7 +670,8 @@ class TestEqualityConvertPKIndex {
// Delete at seq 2 tombstones only the older row; the re-insert (seq 3) survives.
harness.processElement(
- new StreamRecord<>(IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, 2L), 1));
+ new StreamRecord<>(
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, 2L, GLOBAL), 1));
harness.watermark(1);
List<DVPosition> output = harness.extractOutputValues();
@@ -610,7 +680,8 @@ class TestEqualityConvertPKIndex {
// The survivor stays indexed: a later delete with a higher sequence removes it.
harness.processElement(
- new StreamRecord<>(IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, 4L), 2));
+ new StreamRecord<>(
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, 4L, GLOBAL), 2));
harness.watermark(2);
List<DVPosition> afterSecond = harness.extractOutputValues();
@@ -644,7 +715,8 @@ class TestEqualityConvertPKIndex {
false),
0));
harness.processElement(
- new StreamRecord<>(IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, 2L), 1));
+ new StreamRecord<>(
+ IndexCommand.resolveDelete(MAIN_SNAPSHOT, MAIN_SEQ, KEY_1, 2L, GLOBAL), 1));
harness.watermark(1);
List<DVPosition> output = harness.extractOutputValues();
diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestStructLikeSerializer.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestStructLikeSerializer.java
index a740f9432e..8985af325a 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestStructLikeSerializer.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/maintenance/operator/TestStructLikeSerializer.java
@@ -35,8 +35,6 @@ class TestStructLikeSerializer {
Types.NestedField.required(1, "id", Types.IntegerType.get()),
Types.NestedField.optional(2, "data", Types.StringType.get()));
- private static final int SPEC_0 = PartitionSpec.unpartitioned().specId();
-
private final StructLikeSerializer serializer = new StructLikeSerializer();
@Test
@@ -45,8 +43,8 @@ class TestStructLikeSerializer {
key.set(0, 42);
key.set(1, "hello");
- SerializedEqualityValues key1 = serializer.serializeKey(key, KEY_TYPE, SPEC_0);
- SerializedEqualityValues key2 = serializer.serializeKey(key, KEY_TYPE, SPEC_0);
+ SerializedEqualityValues key1 = serializer.serializeKey(key, KEY_TYPE);
+ SerializedEqualityValues key2 = serializer.serializeKey(key, KEY_TYPE);
assertThat(key1).isEqualTo(key2);
}
@@ -60,8 +58,8 @@ class TestStructLikeSerializer {
record2.set(0, 2);
record2.set(1, "b");
- SerializedEqualityValues key1 = serializer.serializeKey(record1, KEY_TYPE, SPEC_0);
- SerializedEqualityValues key2 = serializer.serializeKey(record2, KEY_TYPE, SPEC_0);
+ SerializedEqualityValues key1 = serializer.serializeKey(record1, KEY_TYPE);
+ SerializedEqualityValues key2 = serializer.serializeKey(record2, KEY_TYPE);
assertThat(key1).isNotEqualTo(key2);
}
@@ -78,8 +76,8 @@ class TestStructLikeSerializer {
GenericRecord record3 = GenericRecord.create(typeWithField3);
record3.set(0, 42);
- SerializedEqualityValues key1 = serializer.serializeKey(record1, typeWithField1, SPEC_0);
- SerializedEqualityValues key2 = serializer.serializeKey(record3, typeWithField3, SPEC_0);
+ SerializedEqualityValues key1 = serializer.serializeKey(record1, typeWithField1);
+ SerializedEqualityValues key2 = serializer.serializeKey(record3, typeWithField3);
assertThat(key1).isNotEqualTo(key2);
}
@@ -94,8 +92,8 @@ class TestStructLikeSerializer {
record.set(0, new BigDecimal("123.45"));
record.set(1, "test");
- SerializedEqualityValues key1 = serializer.serializeKey(record, decimalKeyType, SPEC_0);
- SerializedEqualityValues key2 = serializer.serializeKey(record, decimalKeyType, SPEC_0);
+ SerializedEqualityValues key1 = serializer.serializeKey(record, decimalKeyType);
+ SerializedEqualityValues key2 = serializer.serializeKey(record, decimalKeyType);
assertThat(key1).isEqualTo(key2);
}
@@ -105,8 +103,8 @@ class TestStructLikeSerializer {
record.set(0, 1);
record.set(1, null);
- SerializedEqualityValues key1 = serializer.serializeKey(record, KEY_TYPE, SPEC_0);
- SerializedEqualityValues key2 = serializer.serializeKey(record, KEY_TYPE, SPEC_0);
+ SerializedEqualityValues key1 = serializer.serializeKey(record, KEY_TYPE);
+ SerializedEqualityValues key2 = serializer.serializeKey(record, KEY_TYPE);
assertThat(key1).isEqualTo(key2);
}
@@ -120,18 +118,8 @@ class TestStructLikeSerializer {
withValue.set(0, 1);
withValue.set(1, "");
- assertThat(serializer.serializeKey(withNull, KEY_TYPE, SPEC_0))
- .isNotEqualTo(serializer.serializeKey(withValue, KEY_TYPE, SPEC_0));
- }
-
- @Test
- void samePkDifferentSpecsAreNotEqual() {
- GenericRecord pk = GenericRecord.create(KEY_TYPE);
- pk.set(0, 42);
- pk.set(1, "x");
-
- assertThat(serializer.serializeKey(pk, KEY_TYPE, 0))
- .isNotEqualTo(serializer.serializeKey(pk, KEY_TYPE, 1));
+ assertThat(serializer.serializeKey(withNull, KEY_TYPE))
+ .isNotEqualTo(serializer.serializeKey(withValue, KEY_TYPE));
}
@Test