This is an automated email from the ASF dual-hosted git repository.
dybyte pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new d4446df7c9 [Fix][Zeta] Fix IMap file data comparison contract (#11465)
d4446df7c9 is described below
commit d4446df7c979eb109c96258fa87e747e34bd0afb
Author: lifeng <[email protected]>
AuthorDate: Thu Jul 23 00:20:22 2026 +0800
[Fix][Zeta] Fix IMap file data comparison contract (#11465)
---
.../engine/imap/storage/file/bean/IMapData.java | 14 ++-
.../imap/storage/file/bean/IMapDataComparator.java | 104 ++++++++++++++++++
.../imap/storage/file/bean/IMapFileData.java | 14 ++-
.../file/bean/IMapFileDataCompareToTest.java | 119 +++++++++++++++++++++
.../file/common/WALReaderAndWriterTest.java | 37 +++++++
5 files changed, 286 insertions(+), 2 deletions(-)
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapData.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapData.java
index fa2a8731c2..2c812d23b6 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapData.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapData.java
@@ -41,6 +41,18 @@ public class IMapData implements Serializable,
Comparable<IMapData> {
@Override
public int compareTo(IMapData o) {
- return o.timestamp - this.timestamp > 0 ? 1 : -1;
+ return IMapDataComparator.compare(
+ this.deleted,
+ this.key,
+ this.keyClassName,
+ this.value,
+ this.valueClassName,
+ this.timestamp,
+ o.deleted,
+ o.key,
+ o.keyClassName,
+ o.value,
+ o.valueClassName,
+ o.timestamp);
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapDataComparator.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapDataComparator.java
new file mode 100644
index 0000000000..58b2f938cf
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapDataComparator.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.seatunnel.engine.imap.storage.file.bean;
+
+final class IMapDataComparator {
+
+ private IMapDataComparator() {}
+
+ static int compare(
+ boolean leftDeleted,
+ byte[] leftKey,
+ String leftKeyClassName,
+ byte[] leftValue,
+ String leftValueClassName,
+ long leftTimestamp,
+ boolean rightDeleted,
+ byte[] rightKey,
+ String rightKeyClassName,
+ byte[] rightValue,
+ String rightValueClassName,
+ long rightTimestamp) {
+ int keyClassCompare = compareNullableString(leftKeyClassName,
rightKeyClassName);
+ if (keyClassCompare != 0) {
+ return keyClassCompare;
+ }
+
+ int keyCompare = compareBytes(leftKey, rightKey);
+ if (keyCompare != 0) {
+ return keyCompare;
+ }
+
+ int timestampCompare = Long.compare(rightTimestamp, leftTimestamp);
+ if (timestampCompare != 0) {
+ return timestampCompare;
+ }
+
+ // WAL timestamps have millisecond precision and Hadoop FileSystem
does not provide a
+ // portable, atomic global sequence across HDFS, S3, and OSS writers.
These remaining
+ // comparisons therefore provide a deterministic total order for
timestamp collisions;
+ // they do not claim general last-write-wins ordering across
independently active writers.
+ int deleteCompare = Boolean.compare(rightDeleted, leftDeleted);
+ if (deleteCompare != 0) {
+ return deleteCompare;
+ }
+
+ int valueClassCompare = compareNullableString(leftValueClassName,
rightValueClassName);
+ if (valueClassCompare != 0) {
+ return valueClassCompare;
+ }
+
+ return compareBytes(leftValue, rightValue);
+ }
+
+ private static int compareNullableString(String left, String right) {
+ if (left == right) {
+ return 0;
+ }
+ if (left == null) {
+ return -1;
+ }
+ if (right == null) {
+ return 1;
+ }
+ return left.compareTo(right);
+ }
+
+ private static int compareBytes(byte[] left, byte[] right) {
+ if (left == right) {
+ return 0;
+ }
+ if (left == null) {
+ return -1;
+ }
+ if (right == null) {
+ return 1;
+ }
+ int length = Math.min(left.length, right.length);
+ for (int i = 0; i < length; i++) {
+ int compare = Byte.compare(left[i], right[i]);
+ if (compare != 0) {
+ return compare;
+ }
+ }
+ return Integer.compare(left.length, right.length);
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
index 5b9d8863f5..e8920f86fb 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileData.java
@@ -46,6 +46,18 @@ public class IMapFileData implements Serializable,
Comparable<IMapFileData> {
@Override
public int compareTo(IMapFileData o) {
- return o.timestamp - this.timestamp > 0 ? 1 : -1;
+ return IMapDataComparator.compare(
+ this.deleted,
+ this.key,
+ this.keyClassName,
+ this.value,
+ this.valueClassName,
+ this.timestamp,
+ o.deleted,
+ o.key,
+ o.keyClassName,
+ o.value,
+ o.valueClassName,
+ o.timestamp);
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileDataCompareToTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileDataCompareToTest.java
new file mode 100644
index 0000000000..65eca4682c
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/bean/IMapFileDataCompareToTest.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+package org.apache.seatunnel.engine.imap.storage.file.bean;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+public class IMapFileDataCompareToTest {
+
+ @Test
+ public void testCompareToOrdersNewerTimestampFirst() {
+ IMapFileData older = data("key", "value", false, 1000L);
+ IMapFileData newer = data("key", "value", false, 1001L);
+
+ Assertions.assertTrue(newer.compareTo(older) < 0);
+ Assertions.assertTrue(older.compareTo(newer) > 0);
+ }
+
+ @Test
+ public void testCompareToOrdersDeleteBeforePutWithSameTimestamp() {
+ IMapFileData put = data("key", "value", false, 1000L);
+ IMapFileData delete = data("key", null, true, 1000L);
+
+ Assertions.assertTrue(delete.compareTo(put) < 0);
+ Assertions.assertTrue(put.compareTo(delete) > 0);
+ }
+
+ @Test
+ public void testCompareToGroupsRecordsByKeyBeforeTimestamp() {
+ IMapFileData olderA = data("key-a", "older", false, 1000L);
+ IMapFileData newerA = data("key-a", "newer", false, 2000L);
+ IMapFileData keyB = data("key-b", "value", false, 3000L);
+
+ List<IMapFileData> records = new ArrayList<>();
+ records.add(keyB);
+ records.add(olderA);
+ records.add(newerA);
+ Collections.sort(records);
+
+ Assertions.assertEquals(newerA, records.get(0));
+ Assertions.assertEquals(olderA, records.get(1));
+ Assertions.assertEquals(keyB, records.get(2));
+ }
+
+ @Test
+ public void testCompareToIsAntisymmetricWithEqualTimestamps() {
+ IMapFileData left = data("key-a", "value-a", false, 1000L);
+ IMapFileData right = data("key-b", "value-b", true, 1000L);
+
+ int leftToRight = Integer.signum(left.compareTo(right));
+ int rightToLeft = Integer.signum(right.compareTo(left));
+
+ Assertions.assertEquals(-rightToLeft, leftToRight);
+ }
+
+ @Test
+ public void testCompareToReturnsZeroForSameData() {
+ IMapFileData left = data("key", "value", false, 1000L);
+ IMapFileData right = data("key", "value", false, 1000L);
+
+ Assertions.assertEquals(0, left.compareTo(right));
+ Assertions.assertEquals(0, right.compareTo(left));
+ }
+
+ @Test
+ public void testCollectionsSortHandlesManyEqualTimestamps() {
+ List<IMapFileData> data = new ArrayList<>();
+ Random random = new Random(42);
+ for (int i = 0; i < 4000; i++) {
+ data.add(
+ data(
+ "key-" + i,
+ "value-" + i,
+ random.nextBoolean(),
+ 1000L + random.nextInt(50)));
+ }
+
+ Assertions.assertDoesNotThrow(() -> Collections.sort(data));
+ }
+
+ private static IMapFileData data(String key, String value, boolean
deleted, long timestamp) {
+ return IMapFileData.builder()
+ .deleted(deleted)
+ .key(bytes(key))
+ .keyClassName(String.class.getName())
+ .value(bytes(value))
+ .valueClassName(value == null ? null : String.class.getName())
+ .timestamp(timestamp)
+ .build();
+ }
+
+ private static byte[] bytes(String value) {
+ return value == null ? null : value.getBytes(StandardCharsets.UTF_8);
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
index 388be84836..917d02cf79 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReaderAndWriterTest.java
@@ -48,6 +48,8 @@ public class WALReaderAndWriterTest {
private static FileSystem FS;
private static final Path PARENT_PATH = new Path("/tmp/9/");
+ private static final Path SAME_TIMESTAMP_TOMBSTONE_PATH =
+ new Path("/tmp/imap-wal-same-timestamp-tombstone/");
private static final Serializer SERIALIZER = new ProtoStuffSerializer();
@BeforeAll
@@ -114,9 +116,44 @@ public class WALReaderAndWriterTest {
Assertions.assertNull(result.get("key519"));
}
+ @Test
+ public void testReplayKeepsTombstoneForSameKeyAndTimestamp() throws
Exception {
+ String key = "deleted-key";
+ long timestamp = 1000L;
+
+ try (WALWriter writer =
+ new WALWriter(
+ FS, FileConfiguration.HDFS,
SAME_TIMESTAMP_TOMBSTONE_PATH, SERIALIZER)) {
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize(key))
+ .keyClassName(String.class.getName())
+ .value(SERIALIZER.serialize("value"))
+ .valueClassName(String.class.getName())
+ .deleted(false)
+ .timestamp(timestamp)
+ .build());
+ writer.write(
+ IMapFileData.builder()
+ .key(SERIALIZER.serialize(key))
+ .keyClassName(String.class.getName())
+ .deleted(true)
+ .timestamp(timestamp)
+ .build());
+ }
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+ Map<Object, Object> data =
+ reader.loadAllData(SAME_TIMESTAMP_TOMBSTONE_PATH, new
HashSet<>());
+
+ Assertions.assertFalse(data.containsKey(key));
+
Assertions.assertFalse(reader.loadAllKeys(SAME_TIMESTAMP_TOMBSTONE_PATH).contains(key));
+ }
+
@AfterAll
public static void close() throws IOException {
FS.delete(PARENT_PATH, true);
+ FS.delete(SAME_TIMESTAMP_TOMBSTONE_PATH, true);
FS.close();
}
}
