This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new fbc07327d [core] Introduce Context for LookupStoreFactory to avoid
writing metadata (#2817)
fbc07327d is described below
commit fbc07327d7796e4f1af57646e75f8540ad664e0f
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Jan 30 17:41:12 2024 +0800
[core] Introduce Context for LookupStoreFactory to avoid writing metadata
(#2817)
---
.../benchmark/LookupBloomFilterBenchmark.java | 3 +-
.../apache/paimon/lookup/LookupStoreFactory.java | 5 +-
.../apache/paimon/lookup/LookupStoreWriter.java | 7 +-
.../org/apache/paimon/lookup/hash/HashContext.java | 62 +++++++
.../paimon/lookup/hash/HashLookupStoreFactory.java | 4 +-
.../paimon/lookup/hash/HashLookupStoreReader.java | 129 +++----------
.../paimon/lookup/hash/HashLookupStoreWriter.java | 143 +++++++--------
.../java/org/apache/paimon/utils/BloomFilter.java | 12 +-
.../apache/paimon/utils/FileBasedBloomFilter.java | 6 +-
.../lookup/hash/HashLookupStoreFactoryTest.java | 200 ++++++++++-----------
.../apache/paimon/mergetree/ContainsLevels.java | 12 +-
.../org/apache/paimon/mergetree/LookupLevels.java | 12 +-
12 files changed, 276 insertions(+), 319 deletions(-)
diff --git
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java
index 5b5ea766e..e878301d2 100644
---
a/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java
+++
b/paimon-benchmark/paimon-micro-benchmarks/src/test/java/org/apache/paimon/benchmark/LookupBloomFilterBenchmark.java
@@ -137,7 +137,6 @@ public class LookupBloomFilterBenchmark {
for (byte[] input : inputs) {
writer.put(input, value);
}
- writer.close();
- return factory.createReader(file);
+ return factory.createReader(file, writer.close());
}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java
index cde93fef0..c22b8605a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreFactory.java
@@ -42,7 +42,7 @@ public interface LookupStoreFactory {
LookupStoreWriter createWriter(File file, @Nullable BloomFilter.Builder
bloomFilter)
throws IOException;
- LookupStoreReader createReader(File file) throws IOException;
+ LookupStoreReader createReader(File file, Context context) throws
IOException;
static Function<Long, BloomFilter.Builder> bfGenerator(Options options) {
Function<Long, BloomFilter.Builder> bfGenerator = rowCount -> null;
@@ -58,4 +58,7 @@ public interface LookupStoreFactory {
}
return bfGenerator;
}
+
+ /** Context between writer and reader. */
+ interface Context {}
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreWriter.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreWriter.java
index 607681204..60271ce40 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/LookupStoreWriter.java
@@ -18,12 +18,15 @@
package org.apache.paimon.lookup;
-import java.io.Closeable;
+import org.apache.paimon.lookup.LookupStoreFactory.Context;
+
import java.io.IOException;
/** Writer to prepare binary file. */
-public interface LookupStoreWriter extends Closeable {
+public interface LookupStoreWriter {
/** Put key value to store. */
void put(byte[] key, byte[] value) throws IOException;
+
+ Context close() throws IOException;
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashContext.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashContext.java
new file mode 100644
index 000000000..d7d90badd
--- /dev/null
+++ b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashContext.java
@@ -0,0 +1,62 @@
+/*
+ * 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.paimon.lookup.hash;
+
+import org.apache.paimon.lookup.LookupStoreFactory.Context;
+
+/** Context for {@link HashLookupStoreFactory}. */
+public class HashContext implements Context {
+
+ // is bloom filter enabled
+ final boolean bloomFilterEnabled;
+ // expected entries for bloom filter
+ final long bloomFilterExpectedEntries;
+ // bytes for bloom filter
+ final int bloomFilterBytes;
+
+ // Key count for each key length
+ final int[] keyCounts;
+ // Slot size for each key length
+ final int[] slotSizes;
+ // Number of slots for each key length
+ final int[] slots;
+ // Offset of the index for different key length
+ final int[] indexOffsets;
+ // Offset of the data for different key length
+ final long[] dataOffsets;
+
+ public HashContext(
+ boolean bloomFilterEnabled,
+ long bloomFilterExpectedEntries,
+ int bloomFilterBytes,
+ int[] keyCounts,
+ int[] slotSizes,
+ int[] slots,
+ int[] indexOffsets,
+ long[] dataOffsets) {
+ this.bloomFilterEnabled = bloomFilterEnabled;
+ this.bloomFilterExpectedEntries = bloomFilterExpectedEntries;
+ this.bloomFilterBytes = bloomFilterBytes;
+ this.keyCounts = keyCounts;
+ this.slotSizes = slotSizes;
+ this.slots = slots;
+ this.indexOffsets = indexOffsets;
+ this.dataOffsets = dataOffsets;
+ }
+}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java
index e34e97a4e..c4dbcffb9 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreFactory.java
@@ -41,8 +41,8 @@ public class HashLookupStoreFactory implements
LookupStoreFactory {
}
@Override
- public HashLookupStoreReader createReader(File file) throws IOException {
- return new HashLookupStoreReader(cacheManager, cachePageSize, file);
+ public HashLookupStoreReader createReader(File file, Context context)
throws IOException {
+ return new HashLookupStoreReader(file, (HashContext) context,
cacheManager, cachePageSize);
}
@Override
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
index aae039f31..19849953a 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreReader.java
@@ -30,16 +30,10 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-import java.io.BufferedInputStream;
-import java.io.DataInputStream;
import java.io.File;
-import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
-import java.text.DecimalFormat;
-import java.text.SimpleDateFormat;
import java.util.Arrays;
-import java.util.Calendar;
import java.util.Iterator;
import java.util.Map;
@@ -64,111 +58,41 @@ public class HashLookupStoreReader
private FileBasedRandomInputView inputView;
// Buffers
private final byte[] slotBuffer;
+
@Nullable private FileBasedBloomFilter bloomFilter;
- HashLookupStoreReader(CacheManager cacheManager, int cachePageSize, File
file)
+ HashLookupStoreReader(
+ File file, HashContext context, CacheManager cacheManager, int
cachePageSize)
throws IOException {
// File path
if (!file.exists()) {
throw new FileNotFoundException("File " + file.getAbsolutePath() +
" not found");
}
+
+ keyCounts = context.keyCounts;
+ slots = context.slots;
+ slotSizes = context.slotSizes;
+ int maxSlotSize = 0;
+ for (int slotSize : slotSizes) {
+ maxSlotSize = Math.max(maxSlotSize, slotSize);
+ }
+ slotBuffer = new byte[maxSlotSize];
+ indexOffsets = context.indexOffsets;
+ dataOffsets = context.dataOffsets;
+
LOG.info("Opening file {}", file.getName());
// Create Mapped file in read-only mode
inputView = new FileBasedRandomInputView(file, cacheManager,
cachePageSize);
- // Open file and read metadata
- long createdAt;
- FileInputStream inputStream = new FileInputStream(file);
- DataInputStream dataInputStream = new DataInputStream(new
BufferedInputStream(inputStream));
- // Offset of the index in the channel
- int keyCount;
- int indexOffset;
- long dataOffset;
- boolean bloomFilterEnabled;
- try {
- // Time
- createdAt = dataInputStream.readLong();
-
- // Metadata counters
- keyCount = dataInputStream.readInt();
- // Number of different key length
- int keyLengthCount = dataInputStream.readInt();
- // Max key length
- int maxKeyLength = dataInputStream.readInt();
- // BloomFilter
- bloomFilterEnabled = dataInputStream.readBoolean();
- if (bloomFilterEnabled) {
- long recordCount = dataInputStream.readLong();
- int bfBytes = dataInputStream.readInt();
- // Bloom filter buffer start index
- int bfStartIndex = dataInputStream.readInt();
- // Skip bloom filter buffers
- dataInputStream.skipBytes(bfBytes);
- bloomFilter =
- new FileBasedBloomFilter(
- inputView.file(), cacheManager, recordCount,
bfStartIndex, bfBytes);
- }
-
- // Read offset counts and keys
- indexOffsets = new int[maxKeyLength + 1];
- dataOffsets = new long[maxKeyLength + 1];
- keyCounts = new int[maxKeyLength + 1];
- slots = new int[maxKeyLength + 1];
- slotSizes = new int[maxKeyLength + 1];
-
- int maxSlotSize = 0;
- for (int i = 0; i < keyLengthCount; i++) {
- int keyLength = dataInputStream.readInt();
-
- keyCounts[keyLength] = dataInputStream.readInt();
- slots[keyLength] = dataInputStream.readInt();
- slotSizes[keyLength] = dataInputStream.readInt();
- indexOffsets[keyLength] = dataInputStream.readInt();
- dataOffsets[keyLength] = dataInputStream.readLong();
-
- maxSlotSize = Math.max(maxSlotSize, slotSizes[keyLength]);
- }
-
- slotBuffer = new byte[maxSlotSize];
-
- // Read index offset to resign indexOffsets
- indexOffset = dataInputStream.readInt();
- for (int i = 0; i < indexOffsets.length; i++) {
- indexOffsets[i] = indexOffset + indexOffsets[i];
- }
- // Read data offset to resign dataOffsets
- dataOffset = dataInputStream.readLong();
- for (int i = 0; i < dataOffsets.length; i++) {
- dataOffsets[i] = dataOffset + dataOffsets[i];
- }
- } finally {
- // Close metadata
- dataInputStream.close();
- inputStream.close();
+ if (context.bloomFilterEnabled) {
+ bloomFilter =
+ new FileBasedBloomFilter(
+ inputView.file(),
+ cacheManager,
+ context.bloomFilterExpectedEntries,
+ 0,
+ context.bloomFilterBytes);
}
-
- // logging
- DecimalFormat integerFormat = new DecimalFormat("#,##0.00");
- StringBuilder statMsg = new StringBuilder("Storage metadata\n");
- statMsg.append(" Created at:
").append(formatCreatedAt(createdAt)).append("\n");
- statMsg.append(" Key count: ").append(keyCount).append("\n");
- statMsg.append(" Bloom filter:
").append(bloomFilterEnabled).append("\n");
- for (int i = 0; i < keyCounts.length; i++) {
- if (keyCounts[i] > 0) {
- statMsg.append(" Key count for key length ")
- .append(i)
- .append(": ")
- .append(keyCounts[i])
- .append("\n");
- }
- }
- statMsg.append(" Index size: ")
- .append(integerFormat.format((dataOffset - indexOffset) /
(1024.0 * 1024.0)))
- .append(" Mb\n");
- statMsg.append(" Data size: ")
- .append(integerFormat.format((file.length() - dataOffset) /
(1024.0 * 1024.0)))
- .append(" Mb\n");
- LOG.info(statMsg.toString());
}
@Override
@@ -226,13 +150,6 @@ public class HashLookupStoreReader
return res;
}
- private String formatCreatedAt(long createdAt) {
- SimpleDateFormat sdf = new SimpleDateFormat("yyyy.MM.dd G 'at'
HH:mm:ss z");
- Calendar cl = Calendar.getInstance();
- cl.setTimeInMillis(createdAt);
- return sdf.format(cl.getTime());
- }
-
@Override
public void close() throws IOException {
inputView.close();
diff --git
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java
index 972066643..d0d00ba0e 100644
---
a/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/lookup/hash/HashLookupStoreWriter.java
@@ -18,6 +18,7 @@
package org.apache.paimon.lookup.hash;
+import org.apache.paimon.lookup.LookupStoreFactory.Context;
import org.apache.paimon.lookup.LookupStoreWriter;
import org.apache.paimon.utils.BloomFilter;
import org.apache.paimon.utils.MurmurHashUtils;
@@ -68,8 +69,6 @@ public class HashLookupStoreWriter implements
LookupStoreWriter {
private int[] lastValuesLength;
// Data length
private long[] dataLengths;
- // Index length
- private long indexesLength;
// Max offset length
private int[] maxOffsetLengths;
// Number of keys
@@ -79,6 +78,7 @@ public class HashLookupStoreWriter implements
LookupStoreWriter {
private int valueCount;
// Number of collisions
private int collisions;
+
@Nullable private final BloomFilter.Builder bloomFilter;
HashLookupStoreWriter(double loadFactor, File file, @Nullable
BloomFilter.Builder bloomFilter)
@@ -157,7 +157,7 @@ public class HashLookupStoreWriter implements
LookupStoreWriter {
}
@Override
- public void close() throws IOException {
+ public Context close() throws IOException {
// Close the data and index streams
for (DataOutputStream dos : dataStreams) {
if (dos != null) {
@@ -177,17 +177,63 @@ public class HashLookupStoreWriter implements
LookupStoreWriter {
// Prepare files to merge
List<File> filesToMerge = new ArrayList<>();
- try {
+ int bloomFilterBytes = bloomFilter == null ? 0 :
bloomFilter.getBuffer().size();
+ HashContext context =
+ new HashContext(
+ bloomFilter != null,
+ bloomFilter == null ? 0 :
bloomFilter.expectedEntries(),
+ bloomFilterBytes,
+ new int[keyCounts.length],
+ new int[keyCounts.length],
+ new int[keyCounts.length],
+ new int[keyCounts.length],
+ new long[keyCounts.length]);
+
+ long indexesLength = bloomFilterBytes;
+ long datasLength = 0;
+ for (int i = 0; i < this.keyCounts.length; i++) {
+ if (this.keyCounts[i] > 0) {
+ // Write the key Count
+ context.keyCounts[i] = keyCounts[i];
+
+ // Write slot count
+ int slots = (int) Math.round(keyCounts[i] / loadFactor);
+ context.slots[i] = slots;
- // Write metadata file
- File metadataFile = new File(tempFolder, "metadata.dat");
- metadataFile.deleteOnExit();
- FileOutputStream metadataOutputStream = new
FileOutputStream(metadataFile);
- DataOutputStream metadataDataOutputStream = new
DataOutputStream(metadataOutputStream);
- writeMetadata(metadataDataOutputStream);
- metadataDataOutputStream.close();
- metadataOutputStream.close();
- filesToMerge.add(metadataFile);
+ // Write slot size
+ int offsetLength = maxOffsetLengths[i];
+ context.slotSizes[i] = i + offsetLength;
+
+ // Write index offset
+ context.indexOffsets[i] = (int) indexesLength;
+
+ // Increment index length
+ indexesLength += (long) (i + offsetLength) * slots;
+
+ // Write data length
+ context.dataOffsets[i] = datasLength;
+
+ // Increment data length
+ datasLength += dataLengths[i];
+ }
+ }
+
+ // adjust data offsets
+ for (int i = 0; i < context.dataOffsets.length; i++) {
+ context.dataOffsets[i] = indexesLength + context.dataOffsets[i];
+ }
+
+ try {
+ // Write bloom filter file
+ if (bloomFilter != null) {
+ File bloomFilterFile = new File(tempFolder, "bloomfilter.dat");
+ bloomFilterFile.deleteOnExit();
+ try (FileOutputStream bfOutputStream = new
FileOutputStream(bloomFilterFile)) {
+ bfOutputStream.write(bloomFilter.getBuffer().getArray());
+ LOG.info("Bloom filter size: {} bytes",
bloomFilter.getBuffer().size());
+ }
+ filesToMerge.add(bloomFilterFile);
+ }
// Build index file
for (int i = 0; i < indexFiles.length; i++) {
@@ -209,82 +255,13 @@ public class HashLookupStoreWriter implements
LookupStoreWriter {
// Merge and write to output
checkFreeDiskSpace(filesToMerge);
mergeFiles(filesToMerge, outputStream);
+ return context;
} finally {
outputStream.close();
cleanup(filesToMerge);
}
}
- private void writeMetadata(DataOutputStream dataOutputStream) throws
IOException {
- // Write time
- dataOutputStream.writeLong(System.currentTimeMillis());
-
- // Prepare
- int keyLengthCount = getNumKeyCount();
- int maxKeyLength = keyCounts.length - 1;
-
- // Write size (number of keys)
- dataOutputStream.writeInt(keyCount);
-
- // Write the number of different key length
- dataOutputStream.writeInt(keyLengthCount);
-
- // Write the max value for keyLength
- dataOutputStream.writeInt(maxKeyLength);
-
- // Write whether the bloom filter is enabled.
- dataOutputStream.writeBoolean(bloomFilter != null);
-
- if (bloomFilter != null) {
- dataOutputStream.writeLong(bloomFilter.getNumRecords());
- dataOutputStream.writeInt(bloomFilter.getBuffer().size());
-
- // Write the bloom filter start index
- dataOutputStream.writeInt(dataOutputStream.size() + Integer.SIZE /
Byte.SIZE);
-
- dataOutputStream.write(bloomFilter.getBuffer().getArray());
- LOG.info("Bloom filter size: {} bytes",
bloomFilter.getBuffer().size());
- }
-
- // For each keyLength
- long datasLength = 0L;
- for (int i = 0; i < keyCounts.length; i++) {
- if (keyCounts[i] > 0) {
- // Write the key length
- dataOutputStream.writeInt(i);
-
- // Write key count
- dataOutputStream.writeInt(keyCounts[i]);
-
- // Write slot count
- int slots = (int) Math.round(keyCounts[i] / loadFactor);
- dataOutputStream.writeInt(slots);
-
- // Write slot size
- int offsetLength = maxOffsetLengths[i];
- dataOutputStream.writeInt(i + offsetLength);
-
- // Write index offset
- dataOutputStream.writeInt((int) indexesLength);
-
- // Increment index length
- indexesLength += (long) (i + offsetLength) * slots;
-
- // Write data length
- dataOutputStream.writeLong(datasLength);
-
- // Increment data length
- datasLength += dataLengths[i];
- }
- }
-
- // Write the position of the index and the data
- int indexOffset =
- dataOutputStream.size() + (Integer.SIZE / Byte.SIZE) +
(Long.SIZE / Byte.SIZE);
- dataOutputStream.writeInt(indexOffset);
- dataOutputStream.writeLong(indexOffset + indexesLength);
- }
-
private File buildIndex(int keyLength) throws IOException {
long count = keyCounts[keyLength];
int slots = (int) Math.round(count / loadFactor);
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
b/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
index d4fee8ee5..3c5d4e3bc 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/BloomFilter.java
@@ -125,13 +125,13 @@ public class BloomFilter {
private final MemorySegment buffer;
private final BloomFilter filter;
- private final long numRecords;
+ private final long expectedEntries;
- Builder(MemorySegment buffer, long numRecords) {
+ Builder(MemorySegment buffer, long expectedEntries) {
this.buffer = buffer;
- this.filter = new BloomFilter(numRecords, buffer.size());
+ this.filter = new BloomFilter(expectedEntries, buffer.size());
filter.setMemorySegment(buffer, 0);
- this.numRecords = numRecords;
+ this.expectedEntries = expectedEntries;
}
public boolean testHash(int hash) {
@@ -146,8 +146,8 @@ public class BloomFilter {
return buffer;
}
- public long getNumRecords() {
- return numRecords;
+ public long expectedEntries() {
+ return expectedEntries;
}
@VisibleForTesting
diff --git
a/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
b/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
index 84fa35cfc..d6ce54bf3 100644
---
a/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/utils/FileBasedBloomFilter.java
@@ -43,13 +43,13 @@ public class FileBasedBloomFilter {
public FileBasedBloomFilter(
RandomAccessFile file,
CacheManager cacheManager,
- long numRecords,
+ long expectedEntries,
long readOffset,
int readLength) {
this.file = file;
this.cacheManager = cacheManager;
- checkArgument(numRecords >= 0);
- this.filter = new BloomFilter(numRecords, readLength);
+ checkArgument(expectedEntries >= 0);
+ this.filter = new BloomFilter(expectedEntries, readLength);
this.readOffset = readOffset;
this.readLength = readLength;
this.accessCount = 0;
diff --git
a/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java
b/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java
index a2633ff36..c7f386950 100644
---
a/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/lookup/hash/HashLookupStoreFactoryTest.java
@@ -24,24 +24,29 @@ package org.apache.paimon.lookup.hash;
import org.apache.paimon.io.DataOutputSerializer;
import org.apache.paimon.io.cache.CacheManager;
+import org.apache.paimon.lookup.LookupStoreFactory.Context;
import org.apache.paimon.options.MemorySize;
+import
org.apache.paimon.testutils.junit.parameterized.ParameterizedTestExtension;
+import org.apache.paimon.testutils.junit.parameterized.Parameters;
import org.apache.paimon.utils.BloomFilter;
import org.apache.paimon.utils.MathUtils;
import org.apache.paimon.utils.VarLengthIntUtils;
import org.apache.commons.math3.random.RandomDataGenerator;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.io.TempDir;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.MethodSource;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
@@ -51,6 +56,7 @@ import java.util.concurrent.ThreadLocalRandom;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for {@link HashLookupStoreFactory}. */
+@ExtendWith(ParameterizedTestExtension.class)
public class HashLookupStoreFactoryTest {
@TempDir Path tempDir;
@@ -58,9 +64,21 @@ public class HashLookupStoreFactoryTest {
private final RandomDataGenerator random = new RandomDataGenerator();
private final int pageSize = 1024;
+ private final boolean enableBloomFilter;
+
private File file;
private HashLookupStoreFactory factory;
+ public HashLookupStoreFactoryTest(List<Object> var) {
+ this.enableBloomFilter = (Boolean) var.get(0);
+ }
+
+ @SuppressWarnings("unused")
+ @Parameters(name = "enableBf-{0}")
+ public static List<List<Object>> getVarSeg() {
+ return Arrays.asList(Collections.singletonList(true),
Collections.singletonList(false));
+ }
+
@BeforeEach
public void setUp() throws IOException {
this.factory =
@@ -72,10 +90,6 @@ public class HashLookupStoreFactoryTest {
}
}
- public static Object[] enableBloomFilter() {
- return new Object[] {false, true};
- }
-
private BloomFilter.Builder createBloomFiler(boolean enabled) {
if (!enabled) {
return null;
@@ -91,46 +105,43 @@ public class HashLookupStoreFactoryTest {
return str.getBytes(StandardCharsets.UTF_8);
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testEmpty(boolean enableBloomFilter) throws IOException {
+ @TestTemplate
+ public void testEmpty() throws IOException {
HashLookupStoreWriter writer =
factory.createWriter(file,
createBloomFiler(enableBloomFilter));
- writer.close();
+ Context context = writer.close();
assertThat(file.exists()).isTrue();
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
assertThat(reader.lookup(toBytes(1))).isNull();
reader.close();
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testOneKey(boolean enableBloomFilter) throws IOException {
+ @TestTemplate
+ public void testOneKey() throws IOException {
HashLookupStoreWriter writer =
factory.createWriter(file,
createBloomFiler(enableBloomFilter));
writer.put(toBytes(1), toBytes("foo"));
- writer.close();
+ Context context = writer.close();
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
assertThat(reader.lookup(toBytes(1))).isEqualTo(toBytes("foo"));
reader.close();
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testTwoFirstKeyLength(boolean enableBloomFilter) throws
IOException {
+ @TestTemplate
+ public void testTwoFirstKeyLength() throws IOException {
int key1 = 1;
int key2 = 245;
// Write
- writeStore(enableBloomFilter, file, new Object[] {key1, key2}, new
Object[] {1, 6});
+ Context context = writeStore(file, new Object[] {key1, key2}, new
Object[] {1, 6});
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
assertThat(reader.lookup(toBytes(key1))).isEqualTo(toBytes(1));
assertThat(reader.lookup((toBytes(key2)))).isEqualTo(toBytes(6));
assertThat(reader.lookup(toBytes(0))).isNull();
@@ -140,17 +151,16 @@ public class HashLookupStoreFactoryTest {
assertThat(reader.lookup(toBytes(1245))).isNull();
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testKeyLengthGap(boolean enableBf) throws IOException {
+ @TestTemplate
+ public void testKeyLengthGap() throws IOException {
int key1 = 1;
int key2 = 2450;
// Write
- writeStore(enableBf, file, new Object[] {key1, key2}, new Object[] {1,
6});
+ Context context = writeStore(file, new Object[] {key1, key2}, new
Object[] {1, 6});
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
assertThat(reader.lookup(toBytes(key1))).isEqualTo(toBytes(1));
assertThat(reader.lookup((toBytes(key2)))).isEqualTo(toBytes(6));
assertThat(reader.lookup(toBytes(0))).isNull();
@@ -162,17 +172,16 @@ public class HashLookupStoreFactoryTest {
assertThat(reader.lookup(toBytes(2454441))).isNull();
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testKeyLengthStartTwo(boolean enableBf) throws IOException {
+ @TestTemplate
+ public void testKeyLengthStartTwo() throws IOException {
int key1 = 245;
int key2 = 2450;
// Write
- writeStore(enableBf, file, new Object[] {key1, key2}, new Object[] {1,
6});
+ Context context = writeStore(file, new Object[] {key1, key2}, new
Object[] {1, 6});
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
assertThat(reader.lookup(toBytes(key1))).isEqualTo(toBytes(1));
assertThat(reader.lookup((toBytes(key2)))).isEqualTo(toBytes(6));
assertThat(reader.lookup(toBytes(6))).isNull();
@@ -183,9 +192,8 @@ public class HashLookupStoreFactoryTest {
assertThat(reader.lookup(toBytes(2454441))).isNull();
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testDataOnTwoBuffers(boolean enableBf) throws IOException {
+ @TestTemplate
+ public void testDataOnTwoBuffers() throws IOException {
Object[] keys = new Object[] {1, 2, 3};
Object[] values =
new Object[] {
@@ -200,18 +208,17 @@ public class HashLookupStoreFactoryTest {
new HashLookupStoreFactory(
new CacheManager(MemorySize.ofMebiBytes(1)), pageSize,
0.75d);
- writeStore(factory, enableBf, file, keys, values);
+ Context context = writeStore(factory, file, keys, values);
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
for (int i = 0; i < keys.length; i++) {
assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i]));
}
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testDataSizeOnTwoBuffers(boolean enableBf) throws IOException {
+ @TestTemplate
+ public void testDataSizeOnTwoBuffers() throws IOException {
Object[] keys = new Object[] {1, 2, 3};
Object[] values =
new Object[] {
@@ -231,71 +238,61 @@ public class HashLookupStoreFactoryTest {
new HashLookupStoreFactory(
new CacheManager(MemorySize.ofMebiBytes(1)), pageSize,
0.75d);
- writeStore(enableBf, file, keys, values);
+ Context context = writeStore(file, keys, values);
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
for (int i = 0; i < keys.length; i++) {
assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i]));
}
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadStringToString(boolean enableBf) throws IOException {
- testReadKeyToString(generateStringKeys(100), enableBf);
+ @TestTemplate
+ public void testReadStringToString() throws IOException {
+ testReadKeyToString(generateStringKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadIntToString(boolean enableBf) throws IOException {
- testReadKeyToString(generateIntKeys(100), enableBf);
+ @TestTemplate
+ public void testReadIntToString() throws IOException {
+ testReadKeyToString(generateIntKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadDoubleToString(boolean enableBf) throws IOException {
- testReadKeyToString(generateDoubleKeys(100), enableBf);
+ @TestTemplate
+ public void testReadDoubleToString() throws IOException {
+ testReadKeyToString(generateDoubleKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadLongToString(boolean enableBf) throws IOException {
- testReadKeyToString(generateLongKeys(100), enableBf);
+ @TestTemplate
+ public void testReadLongToString() throws IOException {
+ testReadKeyToString(generateLongKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadStringToInt(boolean enableBf) throws IOException {
- testReadKeyToInt(generateStringKeys(100), enableBf);
+ @TestTemplate
+ public void testReadStringToInt() throws IOException {
+ testReadKeyToInt(generateStringKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadByteToInt(boolean enableBf) throws IOException {
- testReadKeyToInt(generateByteKeys(100), enableBf);
+ @TestTemplate
+ public void testReadByteToInt() throws IOException {
+ testReadKeyToInt(generateByteKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadIntToInt(boolean enableBf) throws IOException {
- testReadKeyToInt(generateIntKeys(100), enableBf);
+ @TestTemplate
+ public void testReadIntToInt() throws IOException {
+ testReadKeyToInt(generateIntKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadCompoundToString(boolean enableBf) throws IOException {
- testReadKeyToString(generateCompoundKeys(100), enableBf);
+ @TestTemplate
+ public void testReadCompoundToString() throws IOException {
+ testReadKeyToString(generateCompoundKeys(100));
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testReadCompoundByteToString(boolean enableBf) throws
IOException {
- testReadKeyToString(new Object[] {generateCompoundByteKey()},
enableBf);
+ @TestTemplate
+ public void testReadCompoundByteToString() throws IOException {
+ testReadKeyToString(new Object[] {generateCompoundByteKey()});
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testCacheExpiration(boolean enableBf) throws IOException {
+ @TestTemplate
+ public void testCacheExpiration() throws IOException {
int len = 1000;
ThreadLocalRandom rnd = ThreadLocalRandom.current();
Object[] keys = new Object[len];
@@ -306,32 +303,31 @@ public class HashLookupStoreFactoryTest {
}
// Write
- writeStore(enableBf, file, keys, values);
+ Context context = writeStore(file, keys, values);
// Read
factory =
new HashLookupStoreFactory(new CacheManager(new
MemorySize(8096)), pageSize, 0.75d);
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
for (int i = 0; i < keys.length; i++) {
assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i]));
}
}
- @ParameterizedTest
- @MethodSource(value = "enableBloomFilter")
- public void testIterate(boolean enableBf) throws IOException {
+ @TestTemplate
+ public void testIterate() throws IOException {
Integer[] keys = generateIntKeys(100);
String[] values = generateStringData(keys.length, 12);
// Write
- writeStore(enableBf, file, keys, values);
+ Context context = writeStore(file, keys, values);
// Sets
Set<Integer> keysSet = new HashSet<>(Arrays.asList(keys));
Set<String> valuesSet = new HashSet<>(Arrays.asList(values));
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
Iterator<Map.Entry<byte[], byte[]>> itr = reader.iterator();
for (int i = 0; i < keys.length; i++) {
assertThat(itr.hasNext()).isTrue();
@@ -351,13 +347,13 @@ public class HashLookupStoreFactoryTest {
// UTILITY
- private void testReadKeyToString(Object[] keys, boolean enableBf) throws
IOException {
+ private void testReadKeyToString(Object[] keys) throws IOException {
// Write
Object[] values = generateStringData(keys.length, 10);
- writeStore(enableBf, file, keys, values);
+ Context context = writeStore(file, keys, values);
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
for (int i = 0; i < keys.length; i++) {
assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i]));
@@ -366,13 +362,13 @@ public class HashLookupStoreFactoryTest {
reader.close();
}
- private void testReadKeyToInt(Object[] keys, boolean enableBf) throws
IOException {
+ private void testReadKeyToInt(Object[] keys) throws IOException {
// Write
Integer[] values = generateIntData(keys.length);
- writeStore(enableBf, file, keys, values);
+ Context context = writeStore(file, keys, values);
// Read
- HashLookupStoreReader reader = factory.createReader(file);
+ HashLookupStoreReader reader = factory.createReader(file, context);
for (int i = 0; i < keys.length; i++) {
assertThat(reader.lookup(toBytes(keys[i]))).isEqualTo(toBytes(values[i]));
@@ -381,23 +377,19 @@ public class HashLookupStoreFactoryTest {
reader.close();
}
- private void writeStore(boolean enableBf, File location, Object[] keys,
Object[] values)
- throws IOException {
- writeStore(factory, enableBf, location, keys, values);
+ private Context writeStore(File location, Object[] keys, Object[] values)
throws IOException {
+ return writeStore(factory, location, keys, values);
}
- private void writeStore(
- HashLookupStoreFactory factory,
- boolean enableBf,
- File location,
- Object[] keys,
- Object[] values)
+ private Context writeStore(
+ HashLookupStoreFactory factory, File location, Object[] keys,
Object[] values)
throws IOException {
- HashLookupStoreWriter writer = factory.createWriter(location,
createBloomFiler(enableBf));
+ HashLookupStoreWriter writer =
+ factory.createWriter(location,
createBloomFiler(enableBloomFilter));
for (int i = 0; i < keys.length; i++) {
writer.put(toBytes(keys[i]), toBytes(values[i]));
}
- writer.close();
+ return writer.close();
}
private Integer[] generateIntKeys(int count) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java
index 34f9947ba..cce18363c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/ContainsLevels.java
@@ -154,10 +154,10 @@ public class ContainsLevels implements
Levels.DropFileCallback, Closeable {
if (!localFile.createNewFile()) {
throw new IOException("Can not create new file: " + localFile);
}
- try (LookupStoreWriter kvWriter =
- lookupStoreFactory.createWriter(
- localFile, bfGenerator.apply(file.rowCount()));
- RecordReader<KeyValue> reader = fileReaderFactory.apply(file))
{
+ LookupStoreWriter kvWriter =
+ lookupStoreFactory.createWriter(localFile,
bfGenerator.apply(file.rowCount()));
+ LookupStoreFactory.Context context;
+ try (RecordReader<KeyValue> reader = fileReaderFactory.apply(file)) {
RecordReader.RecordIterator<KeyValue> batch;
KeyValue kv;
while ((batch = reader.readBatch()) != null) {
@@ -170,9 +170,11 @@ public class ContainsLevels implements
Levels.DropFileCallback, Closeable {
} catch (IOException e) {
FileIOUtils.deleteFileOrDirectory(localFile);
throw e;
+ } finally {
+ context = kvWriter.close();
}
- return new ContainsFile(localFile,
lookupStoreFactory.createReader(localFile));
+ return new ContainsFile(localFile,
lookupStoreFactory.createReader(localFile, context));
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
index 9f8b7a196..5420d42a9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java
@@ -171,10 +171,10 @@ public class LookupLevels implements
Levels.DropFileCallback, Closeable {
if (!localFile.createNewFile()) {
throw new IOException("Can not create new file: " + localFile);
}
- try (LookupStoreWriter kvWriter =
- lookupStoreFactory.createWriter(
- localFile, bfGenerator.apply(file.rowCount()));
- RecordReader<KeyValue> reader = fileReaderFactory.apply(file))
{
+ LookupStoreWriter kvWriter =
+ lookupStoreFactory.createWriter(localFile,
bfGenerator.apply(file.rowCount()));
+ LookupStoreFactory.Context context;
+ try (RecordReader<KeyValue> reader = fileReaderFactory.apply(file)) {
DataOutputSerializer valueOut = new DataOutputSerializer(32);
RecordReader.RecordIterator<KeyValue> batch;
KeyValue kv;
@@ -193,9 +193,11 @@ public class LookupLevels implements
Levels.DropFileCallback, Closeable {
} catch (IOException e) {
FileIOUtils.deleteFileOrDirectory(localFile);
throw e;
+ } finally {
+ context = kvWriter.close();
}
- return new LookupFile(localFile, file,
lookupStoreFactory.createReader(localFile));
+ return new LookupFile(localFile, file,
lookupStoreFactory.createReader(localFile, context));
}
@Override
