This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 6fc3445df7 [Improve][Zeta] Optimize WAL recovery memory usage (#11494)
6fc3445df7 is described below
commit 6fc3445df7b4475b2c82ece7fd80b7e769b920e1
Author: Doyeon Kim <[email protected]>
AuthorDate: Tue Sep 22 23:00:44 2026 +0900
[Improve][Zeta] Optimize WAL recovery memory usage (#11494)
---
.../file/common/LatestMutationAccumulator.java | 101 +++++++++++++++++++++
.../engine/imap/storage/file/common/WALReader.java | 92 +++++++++----------
.../storage/file/wal/reader/DefaultReader.java | 92 +++++++++----------
.../imap/storage/file/wal/reader/IFileReader.java | 18 ++++
.../file/common/LatestMutationAccumulatorTest.java | 64 +++++++++++++
.../file/common/WALReaderAndWriterTest.java | 90 ++++++++++++++++++
6 files changed, 362 insertions(+), 95 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/common/LatestMutationAccumulator.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/LatestMutationAccumulator.java
new file mode 100644
index 0000000000..f8e5a705a4
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/LatestMutationAccumulator.java
@@ -0,0 +1,101 @@
+/*
+ * 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.common;
+
+import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/** Retains only the latest WAL mutation for each serialized key. */
+class LatestMutationAccumulator {
+
+ private final Map<SerializedKey, IMapFileData> latestMutations = new
HashMap<>();
+ private final Set<SerializedKey> searchKeys;
+
+ LatestMutationAccumulator(Set<SerializedKey> searchKeys) {
+ this.searchKeys = searchKeys;
+ }
+
+ void accept(IMapFileData mutation) {
+ SerializedKey serializedKey =
+ SerializedKey.of(mutation.getKey(),
mutation.getKeyClassName());
+ if (searchKeys != null && !searchKeys.contains(serializedKey)) {
+ return;
+ }
+
+ IMapFileData current = latestMutations.get(serializedKey);
+ if (current == null) {
+ latestMutations.put(serializedKey, mutation);
+ return;
+ }
+ if (mutation.compareTo(current) < 0) {
+ latestMutations.put(serializedKey, mutation);
+ }
+ }
+
+ int size() {
+ return latestMutations.size();
+ }
+
+ Collection<IMapFileData> values() {
+ return latestMutations.values();
+ }
+
+ Iterator<IMapFileData> iterator() {
+ return latestMutations.values().iterator();
+ }
+
+ static final class SerializedKey {
+ private final byte[] key;
+ private final String className;
+ private final int hashCode;
+
+ private SerializedKey(byte[] key, String className) {
+ this.key = key;
+ this.className = className;
+ this.hashCode = 31 * Arrays.hashCode(key) +
Objects.hashCode(className);
+ }
+
+ static SerializedKey of(byte[] key, String className) {
+ return new SerializedKey(key, className);
+ }
+
+ @Override
+ public boolean equals(Object object) {
+ if (this == object) {
+ return true;
+ }
+ if (!(object instanceof SerializedKey)) {
+ return false;
+ }
+ SerializedKey that = (SerializedKey) object;
+ return Arrays.equals(key, that.key) && Objects.equals(className,
that.className);
+ }
+
+ @Override
+ public int hashCode() {
+ return hashCode;
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
index f4284b6418..f026e2f940 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/common/WALReader.java
@@ -29,21 +29,21 @@ import
org.apache.seatunnel.engine.imap.storage.file.wal.DiscoveryWalFileFactory
import org.apache.seatunnel.engine.imap.storage.file.wal.reader.IFileReader;
import org.apache.seatunnel.engine.serializer.api.Serializer;
-import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
-import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
import java.util.Set;
+import static
org.apache.seatunnel.engine.imap.storage.file.common.LatestMutationAccumulator.SerializedKey;
+
public class WALReader {
private final Serializer serializer;
- private final IFileReader fileReader;
+ private final IFileReader<IMapFileData> fileReader;
public WALReader(FileSystem fs, FileConfiguration configuration,
Serializer serializer)
throws IOException {
@@ -52,66 +52,60 @@ public class WALReader {
this.fileReader.initialize(fs, serializer);
}
- private List<IMapFileData> readAllData(Path parentPath) throws IOException
{
- return this.fileReader.readAllData(parentPath);
- }
-
public Set<Object> loadAllKeys(Path parentPath) throws IOException {
- List<IMapFileData> allData = readAllData(parentPath);
- if (CollectionUtils.isEmpty(allData)) {
- return new HashSet<>();
- }
- Collections.sort(allData);
- Set<Object> result = new HashSet<>(allData.size());
- Map<Object, Long> deleteMap = new HashMap<>();
- for (IMapFileData data : allData) {
- Object key = deserializeData(data.getKey(),
data.getKeyClassName());
- if (deleteMap.containsKey(key)) {
- continue;
- }
- if (data.isDeleted()) {
- deleteMap.put(key, data.getTimestamp());
- continue;
- }
- if (result.contains(key)) {
- continue;
+ LatestMutationAccumulator accumulator =
readLatestMutations(parentPath, null);
+ Set<Object> result = new HashSet<>(accumulator.size());
+ Iterator<IMapFileData> mutations = accumulator.iterator();
+ while (mutations.hasNext()) {
+ IMapFileData mutation = mutations.next();
+ mutations.remove();
+ if (!mutation.isDeleted()) {
+ result.add(deserializeData(mutation.getKey(),
mutation.getKeyClassName()));
}
- result.add(key);
}
return result;
}
public Map<Object, Object> loadAllData(Path parentPath, Set<Object>
searchKeys)
throws IOException {
- List<IMapFileData> allData = readAllData(parentPath);
- if (CollectionUtils.isEmpty(allData)) {
- return new HashMap<>();
- }
- Collections.sort(allData);
- Map<Object, Object> result = new HashMap<>(allData.size());
- Map<Object, Long> deleteMap = new HashMap<>();
- boolean searchByKeys = CollectionUtils.isNotEmpty(searchKeys);
- for (IMapFileData data : allData) {
- Object key = deserializeData(data.getKey(),
data.getKeyClassName());
- if (searchByKeys && !searchKeys.contains(data.getKey())) {
- continue;
- }
- if (deleteMap.containsKey(key)) {
+ Set<SerializedKey> serializedSearchKeys =
serializeSearchKeys(searchKeys);
+ LatestMutationAccumulator accumulator =
+ readLatestMutations(parentPath, serializedSearchKeys);
+ Map<Object, Object> result = new HashMap<>(accumulator.size());
+ Iterator<IMapFileData> mutations = accumulator.iterator();
+ while (mutations.hasNext()) {
+ IMapFileData mutation = mutations.next();
+ mutations.remove();
+ if (mutation.isDeleted()) {
continue;
}
- if (data.isDeleted()) {
- deleteMap.put(key, data.getTimestamp());
- continue;
- }
- if (result.containsKey(key)) {
- continue;
- }
- Object value = deserializeData(data.getValue(),
data.getValueClassName());
+ Object key = deserializeData(mutation.getKey(),
mutation.getKeyClassName());
+ Object value = deserializeData(mutation.getValue(),
mutation.getValueClassName());
result.put(key, value);
}
return result;
}
+ private LatestMutationAccumulator readLatestMutations(
+ Path parentPath, Set<SerializedKey> serializedSearchKeys) throws
IOException {
+ LatestMutationAccumulator accumulator = new
LatestMutationAccumulator(serializedSearchKeys);
+ fileReader.forEachData(parentPath, accumulator::accept);
+ return accumulator;
+ }
+
+ private Set<SerializedKey> serializeSearchKeys(Set<Object> searchKeys)
throws IOException {
+ if (searchKeys == null || searchKeys.isEmpty()) {
+ return null;
+ }
+
+ Set<SerializedKey> serializedKeys = new HashSet<>(searchKeys.size());
+ for (Object key : searchKeys) {
+ serializedKeys.add(
+ SerializedKey.of(serializer.serialize(key),
key.getClass().getName()));
+ }
+ return serializedKeys;
+ }
+
private Object deserializeData(byte[] data, String className) {
try {
Class<?> clazz = ClassUtils.getClass(className);
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/DefaultReader.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/DefaultReader.java
index f8d5e7fadf..4d900bb4d3 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/DefaultReader.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/DefaultReader.java
@@ -19,18 +19,18 @@
*/
package org.apache.seatunnel.engine.imap.storage.file.wal.reader;
-import
org.apache.seatunnel.engine.imap.storage.api.exception.IMapStorageException;
import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
import org.apache.seatunnel.engine.imap.storage.file.common.WALDataUtils;
import org.apache.seatunnel.engine.serializer.api.Serializer;
-import org.apache.commons.collections.CollectionUtils;
-import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -38,7 +38,6 @@ import java.util.List;
import static
org.apache.seatunnel.engine.imap.storage.file.common.WALDataUtils.WAL_DATA_METADATA_LENGTH;
public class DefaultReader implements IFileReader<IMapFileData> {
- private static final int DEFAULT_QUERY_LIST_SIZE = 1024;
FileSystem fs;
Serializer serializer;
@@ -55,60 +54,61 @@ public class DefaultReader implements
IFileReader<IMapFileData> {
@Override
public List<IMapFileData> readAllData(Path parentPath) throws IOException {
- List<String> fileNames = getFileNames(parentPath);
- if (CollectionUtils.isEmpty(fileNames)) {
- return new ArrayList<>();
- }
- List<IMapFileData> result = new ArrayList<>(DEFAULT_QUERY_LIST_SIZE);
- for (String fileName : fileNames) {
- result.addAll(readData(new Path(parentPath, fileName)));
- }
+ List<IMapFileData> result = new ArrayList<>();
+ forEachData(parentPath, result::add);
return result;
}
- private List<String> getFileNames(Path parentPath) {
- try {
- if (!fs.exists(parentPath)) {
- return new ArrayList<>();
- }
- RemoteIterator<LocatedFileStatus> fileStatusRemoteIterator =
- fs.listFiles(parentPath, true);
- List<String> fileNames = new ArrayList<>();
- while (fileStatusRemoteIterator.hasNext()) {
- LocatedFileStatus fileStatus = fileStatusRemoteIterator.next();
- if (fileStatus.getPath().getName().endsWith("wal.txt")) {
- fileNames.add(fileStatus.getPath().toString());
- }
+ @Override
+ public void forEachData(Path parentPath, RecordConsumer<IMapFileData>
consumer)
+ throws IOException {
+ if (!fs.exists(parentPath)) {
+ return;
+ }
+
+ RemoteIterator<LocatedFileStatus> files = fs.listFiles(parentPath,
true);
+ while (files.hasNext()) {
+ LocatedFileStatus file = files.next();
+ if (file.getPath().getName().endsWith("wal.txt")) {
+ readData(file, consumer);
}
- return fileNames;
- } catch (IOException e) {
- throw new IMapStorageException(e, "get file names error,path is
s%", parentPath);
}
}
- private List<IMapFileData> readData(Path path) throws IOException {
- List<IMapFileData> result = new ArrayList<>(DEFAULT_QUERY_LIST_SIZE);
- long length = fs.getFileStatus(path).getLen();
- try (FSDataInputStream in = fs.open(path)) {
- byte[] datas = new byte[(int) length];
- in.readFully(datas);
- int startIndex = 0;
- while (startIndex + WAL_DATA_METADATA_LENGTH < datas.length) {
+ private void readData(LocatedFileStatus file, RecordConsumer<IMapFileData>
consumer)
+ throws IOException {
+ long remainingBytes = file.getLen();
+ byte[] metadata = new byte[WAL_DATA_METADATA_LENGTH];
+ try (DataInputStream input =
+ new DataInputStream(new
BufferedInputStream(fs.open(file.getPath())))) {
+ while (remainingBytes >= WAL_DATA_METADATA_LENGTH) {
+ if (!readFully(input, metadata)) {
+ break;
+ }
+ remainingBytes -= WAL_DATA_METADATA_LENGTH;
- byte[] metadata = new byte[WAL_DATA_METADATA_LENGTH];
- System.arraycopy(datas, startIndex, metadata, 0,
WAL_DATA_METADATA_LENGTH);
int dataLength = WALDataUtils.byteArrayToInt(metadata);
- startIndex += WAL_DATA_METADATA_LENGTH;
- if (startIndex + dataLength > datas.length) {
+ if (dataLength > remainingBytes) {
+ // The writer may have stopped in the middle of its last
record.
break;
}
- byte[] data = new byte[dataLength];
- System.arraycopy(datas, startIndex, data, 0, data.length);
- IMapFileData fileData = serializer.deserialize(data,
IMapFileData.class);
- result.add(fileData);
- startIndex += data.length;
+
+ byte[] serializedRecord = new byte[dataLength];
+ if (!readFully(input, serializedRecord)) {
+ break;
+ }
+ remainingBytes -= dataLength;
+ consumer.accept(serializer.deserialize(serializedRecord,
IMapFileData.class));
}
}
- return result;
+ }
+
+ private boolean readFully(DataInputStream input, byte[] target) throws
IOException {
+ try {
+ input.readFully(target);
+ return true;
+ } catch (EOFException ignored) {
+ return false;
+ }
}
}
diff --git
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/IFileReader.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/IFileReader.java
index 1fbc85706e..e382a91b0a 100644
---
a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/IFileReader.java
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/main/java/org/apache/seatunnel/engine/imap/storage/file/wal/reader/IFileReader.java
@@ -33,4 +33,22 @@ public interface IFileReader<R> {
void initialize(FileSystem fs, Serializer serializer) throws IOException;
List<R> readAllData(Path parentPath) throws IOException;
+
+ /**
+ * Passes records to the consumer.
+ *
+ * <p>Implementations should override this method to stream records
without retaining the
+ * complete history in memory. The default implementation preserves
compatibility with readers
+ * that only implement {@link #readAllData(Path)}.
+ */
+ default void forEachData(Path parentPath, RecordConsumer<R> consumer)
throws IOException {
+ for (R record : readAllData(parentPath)) {
+ consumer.accept(record);
+ }
+ }
+
+ @FunctionalInterface
+ interface RecordConsumer<R> {
+ void accept(R record) throws IOException;
+ }
}
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/LatestMutationAccumulatorTest.java
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/LatestMutationAccumulatorTest.java
new file mode 100644
index 0000000000..fab89287a1
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/common/LatestMutationAccumulatorTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.common;
+
+import org.apache.seatunnel.engine.imap.storage.file.bean.IMapFileData;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+
+class LatestMutationAccumulatorTest {
+
+ @Test
+ void shouldRetainAtMostOneMutationPerKey() {
+ LatestMutationAccumulator accumulator = new
LatestMutationAccumulator(null);
+
+ for (int version = 0; version < 100_000; version++) {
+ accumulator.accept(mutation("key-" + (version % 100), version,
false));
+ }
+
+ Assertions.assertEquals(100, accumulator.size());
+ Assertions.assertTrue(
+ accumulator.values().stream()
+ .allMatch(mutation -> mutation.getTimestamp() >=
99_900));
+ }
+
+ @Test
+ void shouldChooseNewerMutationRegardlessOfArrivalOrder() {
+ LatestMutationAccumulator accumulator = new
LatestMutationAccumulator(null);
+
+ accumulator.accept(mutation("key", 20, false));
+ accumulator.accept(mutation("key", 10, false));
+
+ Assertions.assertEquals(1, accumulator.size());
+ Assertions.assertEquals(20,
accumulator.values().iterator().next().getTimestamp());
+ }
+
+ private static IMapFileData mutation(String key, long timestamp, boolean
deleted) {
+ return IMapFileData.builder()
+ .key(key.getBytes(StandardCharsets.UTF_8))
+ .keyClassName(String.class.getName())
+ .value(("value-" + timestamp).getBytes(StandardCharsets.UTF_8))
+ .valueClassName(String.class.getName())
+ .timestamp(timestamp)
+ .deleted(deleted)
+ .build();
+ }
+}
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 917d02cf79..6b1fb4b7ff 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
@@ -26,18 +26,23 @@ import
org.apache.seatunnel.engine.serializer.api.Serializer;
import org.apache.seatunnel.engine.serializer.protobuf.ProtoStuffSerializer;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.condition.EnabledOnOs;
import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.condition.OS.LINUX;
@@ -60,6 +65,11 @@ public class WALReaderAndWriterTest {
FS = FileSystem.getLocal(conf);
}
+ @BeforeEach
+ void cleanWalDirectory() throws IOException {
+ FS.delete(PARENT_PATH, true);
+ }
+
@Test
public void testWriterAndReader() throws Exception {
WALWriter writer = new WALWriter(FS, FileConfiguration.HDFS,
PARENT_PATH, SERIALIZER);
@@ -116,6 +126,86 @@ public class WALReaderAndWriterTest {
Assertions.assertNull(result.get("key519"));
}
+ @Test
+ void shouldChooseLatestMutationsAcrossUnsortedFiles() throws IOException {
+ Path path = new Path(PARENT_PATH, "unsorted");
+ writeWalFile(
+ new Path(path, "region-a/older-wal.txt"),
+ mutation("updated", "old", 10, false),
+ mutation("deleted", "old", 10, false));
+ writeWalFile(
+ new Path(path, "region-b/newer-wal.txt"),
+ mutation("updated", "new", 20, false),
+ mutation("deleted", null, 20, true));
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+ Map<Object, Object> result = reader.loadAllData(path,
Collections.emptySet());
+ Set<Object> keys = reader.loadAllKeys(path);
+
+ Assertions.assertEquals(Collections.singletonMap("updated", "new"),
result);
+ Assertions.assertEquals(Collections.singleton("updated"), keys);
+ }
+
+ @Test
+ void shouldRetainOnlyRequestedKeysWhileScanning() throws IOException {
+ Path path = new Path(PARENT_PATH, "selective");
+ writeWalFile(
+ new Path(path, "region/selective-wal.txt"),
+ mutation("key-a", "value-a", 10, false),
+ mutation("key-b", "value-b", 20, false),
+ mutation("key-c", "value-c", 30, false));
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+ Map<Object, Object> result =
+ reader.loadAllData(path, new HashSet<>(Arrays.asList("key-a",
"missing")));
+
+ Assertions.assertEquals(Collections.singletonMap("key-a", "value-a"),
result);
+ }
+
+ @Test
+ void shouldIgnoreIncompleteTrailingRecord() throws IOException {
+ Path path = new Path(PARENT_PATH, "partial");
+ Path file = new Path(path, "region/partial-wal.txt");
+ FS.mkdirs(file.getParent());
+ byte[] complete = wrapped(mutation("complete", "value", 10, false));
+ byte[] incomplete = wrapped(mutation("partial", "ignored", 20, false));
+ try (FSDataOutputStream output = FS.create(file, true)) {
+ output.write(complete);
+ output.write(incomplete, 0, WALDataUtils.WAL_DATA_METADATA_LENGTH
+ 3);
+ }
+
+ WALReader reader = new WALReader(FS, FileConfiguration.HDFS,
SERIALIZER);
+
+ Assertions.assertEquals(
+ Collections.singletonMap("complete", "value"),
+ reader.loadAllData(path, Collections.emptySet()));
+ }
+
+ private static IMapFileData mutation(String key, String value, long
timestamp, boolean deleted)
+ throws IOException {
+ return IMapFileData.builder()
+ .key(SERIALIZER.serialize(key))
+ .keyClassName(String.class.getName())
+ .value(deleted ? null : SERIALIZER.serialize(value))
+ .valueClassName(deleted ? null : String.class.getName())
+ .timestamp(timestamp)
+ .deleted(deleted)
+ .build();
+ }
+
+ private static void writeWalFile(Path file, IMapFileData... mutations)
throws IOException {
+ FS.mkdirs(file.getParent());
+ try (FSDataOutputStream output = FS.create(file, true)) {
+ for (IMapFileData mutation : mutations) {
+ output.write(wrapped(mutation));
+ }
+ }
+ }
+
+ private static byte[] wrapped(IMapFileData mutation) throws IOException {
+ return WALDataUtils.wrapperBytes(SERIALIZER.serialize(mutation));
+ }
+
@Test
public void testReplayKeepsTombstoneForSameKeyAndTimestamp() throws
Exception {
String key = "deleted-key";
