From cb6b147ee46ab4fc98d8a0b2de4aa9098ae4f527 Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Thu, 28 Aug 2025 12:00:10 -0700 Subject: [PATCH 1/9] Make Lucene serialization able to encrypt, using the same serialization key manager as record serialization. --- .../lucene/LuceneRecordContextProperties.java | 3 + .../record/lucene/directory/FDBDirectory.java | 17 +- .../lucene/directory/LuceneSerializer.java | 209 +++++++++++++----- .../record/lucene/FDBLuceneQueryTest.java | 39 +++- ...ionTest.java => LuceneSerializerTest.java} | 43 +++- .../directory/FDBDirectoryFailuresTest.java | 2 +- .../lucene/directory/FDBDirectoryTest.java | 5 +- 7 files changed, 244 insertions(+), 74 deletions(-) rename fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/{LuceneCompressionTest.java => LuceneSerializerTest.java} (66%) diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneRecordContextProperties.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneRecordContextProperties.java index fb4d8f3670..74b8594fa5 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneRecordContextProperties.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneRecordContextProperties.java @@ -21,6 +21,7 @@ package com.apple.foundationdb.record.lucene; import com.apple.foundationdb.record.lucene.directory.FDBDirectory; +import com.apple.foundationdb.record.provider.common.SerializationKeyManager; import com.apple.foundationdb.record.provider.foundationdb.properties.RecordLayerPropertyKey; import com.apple.foundationdb.record.provider.foundationdb.properties.RecordLayerPropertyStorage; @@ -46,6 +47,8 @@ public final class LuceneRecordContextProperties { */ public static final RecordLayerPropertyKey LUCENE_INDEX_ENCRYPTION_ENABLED = RecordLayerPropertyKey.booleanPropertyKey("com.apple.foundationdb.record.lucene.encryptionEnabled", false); + public static final RecordLayerPropertyKey LUCENE_INDEX_KEY_MANAGER = new RecordLayerPropertyKey<>("com.apple.foundationdb.record.lucene.keyManager", null, SerializationKeyManager.class); + /** * An {@link ExecutorService} to use for parallel execution in {@link LuceneRecordCursor}. */ diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java index fd5725c135..b5971b0f89 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java @@ -181,6 +181,7 @@ public class FDBDirectory extends Directory { // True if sharedCacheManager is present until sharedCache has been set (or not). private boolean sharedCachePending; private final AgilityContext agilityContext; + private final LuceneSerializer serializer; @Nullable private LucenePrimaryKeySegmentIndex primaryKeySegmentIndex; @@ -231,6 +232,7 @@ private FDBDirectory(@Nonnull Subspace subspace, @Nullable Map i this.fileSequenceCounter = new AtomicLong(-1); this.compressionEnabled = Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED), false); this.encryptionEnabled = Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_ENCRYPTION_ENABLED), false); + this.serializer = new LuceneSerializer(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER)); this.fileReferenceMapSupplier = Suppliers.memoize(this::loadFileReferenceCacheForMemoization); this.sharedCacheManager = sharedCacheManager; this.sharedCacheKey = sharedCacheKey; @@ -403,7 +405,7 @@ public static boolean isStoredFieldsFile(String name) { */ public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLuceneFileReference reference) { final byte[] fileReferenceBytes = reference.getBytes(); - final byte[] encodedBytes = Objects.requireNonNull(LuceneSerializer.encode(fileReferenceBytes, compressionEnabled, encryptionEnabled)); + final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(fileReferenceBytes, compressionEnabled, encryptionEnabled)); agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE_FILE_REFERENCE, encodedBytes.length); if (LOGGER.isTraceEnabled()) { LOGGER.trace(getLogMessage("Write lucene file reference", @@ -425,7 +427,7 @@ public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLucene * @return the actual data size written to database with potential compression and encryption applied */ public int writeData(final long id, final int block, @Nonnull final byte[] value) { - final byte[] encodedBytes = Objects.requireNonNull(LuceneSerializer.encode(value, compressionEnabled, encryptionEnabled)); + final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(value, compressionEnabled, encryptionEnabled)); //This may not be correct transactionally agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE, encodedBytes.length); if (LOGGER.isTraceEnabled()) { @@ -538,7 +540,7 @@ private CompletableFuture readBlock(@Nonnull IndexInput requestingInput, private CompletableFuture readData(long id, int block) { return agilityContext.instrument(LuceneEvents.Events.LUCENE_FDB_READ_BLOCK, agilityContext.get(dataSubspace.pack(Tuple.from(id, block))) - .thenApply(LuceneSerializer::decode)); + .thenApply(serializer::decode)); } @Nonnull @@ -624,7 +626,7 @@ private CompletableFuture loadFileReferenceCacheForMemoization() { agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_FILES_COUNT, list.size()); list.forEach(kv -> { String name = metaSubspace.unpack(kv.getKey()).getString(0); - final FDBLuceneFileReference fileReference = Objects.requireNonNull(FDBLuceneFileReference.parseFromBytes(LuceneSerializer.decode(kv.getValue()))); + final FDBLuceneFileReference fileReference = Objects.requireNonNull(FDBLuceneFileReference.parseFromBytes(serializer.decode(kv.getValue()))); outMap.put(name, fileReference); if (fileReference.getFieldInfosId() != 0) { fieldInfosCount.computeIfAbsent(fileReference.getFieldInfosId(), key -> new AtomicInteger(0)) @@ -915,7 +917,7 @@ public void rename(@Nonnull final String source, @Nonnull final String dest) thr .addLogInfo(LuceneLogMessageKeys.COMPRESSION_SUPPOSED, compressionEnabled) .addLogInfo(LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, encryptionEnabled); } - byte[] encodedBytes = LuceneSerializer.encode(value.getBytes(), compressionEnabled, encryptionEnabled); + byte[] encodedBytes = serializer.encode(value.getBytes(), compressionEnabled, encryptionEnabled); agilityContext.set(metaSubspace.pack(dest), encodedBytes); agilityContext.clear(key); @@ -1037,6 +1039,11 @@ public T asyncToSync(@Nonnull StoreTimer.Wait event, @Nonnull CompletableFut return agilityContext.asyncToSync(event, async); } + @Nullable + public LuceneSerializer getSerializer() { + return serializer; + } + public Subspace getSubspace() { return subspace; } diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java index bd22ad2a85..80df13283b 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java @@ -22,8 +22,9 @@ import com.apple.foundationdb.record.RecordCoreException; import com.apple.foundationdb.record.logging.KeyValueLogMessage; -import com.apple.foundationdb.record.logging.LogMessageKeys; import com.apple.foundationdb.record.lucene.LuceneLogMessageKeys; +import com.apple.foundationdb.record.provider.common.CipherPool; +import com.apple.foundationdb.record.provider.common.SerializationKeyManager; import org.apache.lucene.codecs.compressing.CompressionMode; import org.apache.lucene.codecs.compressing.Compressor; import org.apache.lucene.codecs.compressing.Decompressor; @@ -35,7 +36,10 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; -import java.util.Arrays; +import javax.crypto.Cipher; +import javax.crypto.spec.IvParameterSpec; +import java.io.IOException; +import java.security.GeneralSecurityException; /** * Serialize a Lucene directory block to/from an FDB key-value byte array. @@ -46,55 +50,85 @@ public class LuceneSerializer { private static final int ENCODING_ENCRYPTED = 1; private static final int ENCODING_COMPRESSED = 2; private static final byte COMPRESSION_VERSION_FOR_HIGH_COMPRESSION = 0; + private static final int ENCRYPTION_KEY_SHIFT = 3; + + @Nullable + private final SerializationKeyManager keyManager; + + public LuceneSerializer(@Nullable SerializationKeyManager keyManager) { + this.keyManager = keyManager; + } private static class EncodingState { private boolean compressed; private boolean encrypted; + private int keyNumber; private EncodingState() { this.compressed = false; this.encrypted = false; } + boolean isCompressed() { + return compressed; + } + void setCompressed(boolean compressed) { this.compressed = compressed; } + boolean isEncrypted() { + return encrypted; + } + void setEncrypted(boolean encrypted) { this.encrypted = encrypted; } - boolean isCompressed() { - return compressed; + public int getKeyNumber() { + return keyNumber; } - boolean isEncrypted() { - return encrypted; + public void setKeyNumber(final int keyNumber) { + this.keyNumber = keyNumber; } } @Nullable - public static byte[] encode(@Nullable byte[] data, boolean compress, boolean encrypt) { + public byte[] encode(@Nullable byte[] data, boolean compress, boolean encrypt) { if (data == null) { return null; } final EncodingState state = new EncodingState(); - final ByteBuffersDataOutput decodedDataOutput = new ByteBuffersDataOutput(); - // Placeholder for the code byte at beginning, will be modified in output byte array if needed - decodedDataOutput.writeByte((byte) 0); - - byte[] encoded = compressIfNeeded(compress, data, decodedDataOutput, state, 1); - - int code = 0; - if (state.isCompressed()) { - code = code | ENCODING_COMPRESSED; + long prefix = 0; + if (compress) { + prefix |= ENCODING_COMPRESSED; + state.setCompressed(true); } // Encryption will be supported in future - if (state.isEncrypted()) { - code = code | ENCODING_ENCRYPTED; + if (encrypt) { + if (keyManager == null) { + throw new RecordCoreException("cannot encrypt Lucene blocks without keys"); + } + int key = keyManager.getSerializationKey(); + prefix |= ENCODING_ENCRYPTED | ((key & 0xFFFFFFFFL) << ENCRYPTION_KEY_SHIFT); + state.setEncrypted(true); + state.setKeyNumber(key); + } + + byte[] encoded; + try { + final ByteBuffersDataOutput decodedDataOutput = new ByteBuffersDataOutput(); + // Placeholder for the code byte at beginning, will be modified in output byte array if needed + decodedDataOutput.writeVLong(prefix); + final int prefixLength = (int)decodedDataOutput.size(); + encoded = compressIfNeeded(state, decodedDataOutput, data, prefixLength); + encoded = encryptIfNeeded(state, encoded, prefixLength); + } catch (IOException | GeneralSecurityException ex) { + throw new RecordCoreException("Lucene data encoding failure", ex); } - encoded[0] = (byte) code; + if (LOGGER.isTraceEnabled()) { LOGGER.trace(KeyValueLogMessage.of("Encoded lucene data", LuceneLogMessageKeys.COMPRESSION_SUPPOSED, compress, @@ -104,11 +138,12 @@ public static byte[] encode(@Nullable byte[] data, boolean compress, boolean enc LuceneLogMessageKeys.ORIGINAL_DATA_SIZE, data.length, LuceneLogMessageKeys.ENCODED_DATA_SIZE, encoded.length)); } + return encoded; } @Nullable - public static byte[] decode(@Nullable byte[] data) { + public byte[] decode(@Nullable byte[] data) { if (data == null) { return null; } @@ -118,81 +153,137 @@ public static byte[] decode(@Nullable byte[] data) { .addLogInfo(LuceneLogMessageKeys.DATA_VALUE, data); } - final byte encoding = data[0]; - final boolean encrypted = (encoding & ENCODING_ENCRYPTED) == ENCODING_ENCRYPTED; - final boolean compressed = (encoding & ENCODING_COMPRESSED) == ENCODING_COMPRESSED; + final EncodingState state = new EncodingState(); + byte[] decoded; + try { + final ByteArrayDataInput encodedDataInput = new ByteArrayDataInput(data); + final long prefix = encodedDataInput.readVLong(); + + state.setCompressed((prefix & ENCODING_COMPRESSED) != 0); + state.setEncrypted((prefix & ENCODING_ENCRYPTED) != 0); + state.setKeyNumber((int)(prefix >> ENCRYPTION_KEY_SHIFT)); + + decryptIfNeeded(state, encodedDataInput); + decompressIfNeeded(state, encodedDataInput); + + decoded = new byte[encodedDataInput.length() - encodedDataInput.getPosition()]; + encodedDataInput.readBytes(decoded, 0, decoded.length); + } catch (IOException | GeneralSecurityException ex) { + throw new RecordCoreException("Lucene data decoding failure", ex); + } - byte[] decoded = decompressIfNeeded(compressed, data, 1); if (LOGGER.isTraceEnabled()) { LOGGER.trace(KeyValueLogMessage.of("Decoded lucene data", - LuceneLogMessageKeys.COMPRESSED_EVENTUALLY, compressed, - LuceneLogMessageKeys.ENCRYPTED_EVENTUALLY, encrypted, + LuceneLogMessageKeys.COMPRESSED_EVENTUALLY, state.isCompressed(), + LuceneLogMessageKeys.ENCRYPTED_EVENTUALLY, state.isCompressed(), LuceneLogMessageKeys.ENCODED_DATA_SIZE, data.length, LuceneLogMessageKeys.ORIGINAL_DATA_SIZE, decoded.length)); } + return decoded; } @Nonnull - private static byte[] compressIfNeeded(boolean compressionNeeded, @Nonnull byte[] data, @Nonnull ByteBuffersDataOutput encodedDataOutput, - @Nonnull EncodingState state, int offset) { - if (!compressionNeeded) { - return fallBackToUncompressed(encodedDataOutput.toArrayCopy(), data, state, offset); + private static byte[] compressIfNeeded(@Nonnull EncodingState state, @Nonnull ByteBuffersDataOutput encodedDataOutput, + @Nonnull byte[] uncompressedData, int prefixLength) + throws IOException { + if (!state.isCompressed()) { + return fallBackToUncompressed(state, uncompressedData, encodedDataOutput.toArrayCopy(), prefixLength); } try (Compressor compressor = CompressionMode.HIGH_COMPRESSION.newCompressor()) { encodedDataOutput.writeByte(COMPRESSION_VERSION_FOR_HIGH_COMPRESSION); - encodedDataOutput.writeVInt(data.length); - compressor.compress(data, 0, data.length, encodedDataOutput); + encodedDataOutput.writeVInt(uncompressedData.length); + compressor.compress(uncompressedData, 0, uncompressedData.length, encodedDataOutput); final byte[] compressedData = encodedDataOutput.toArrayCopy(); // Compress only if it helps to shorten the bytes - if (compressedData.length < data.length) { + if (compressedData.length < uncompressedData.length) { state.setCompressed(true); return compressedData; } else { - return fallBackToUncompressed(compressedData, data, state, offset); + return fallBackToUncompressed(state, uncompressedData, compressedData, prefixLength); } - } catch (Exception e) { - throw new RecordCoreException("Lucene data compression failure", e); } } - private static byte[] fallBackToUncompressed(@Nonnull byte[] compressedData, @Nonnull byte[] originalData, - @Nonnull EncodingState state, int offset) { + private static byte[] fallBackToUncompressed(@Nonnull EncodingState state, @Nonnull byte[] originalData, + @Nonnull byte[] encodedData, int prefixLength) { + final byte[] encoded = new byte[originalData.length + prefixLength]; + System.arraycopy(encodedData, 0, encoded, 0, prefixLength); + encoded[0] &= ~ENCODING_COMPRESSED; + System.arraycopy(originalData, 0, encoded, prefixLength, originalData.length); state.setCompressed(false); - final byte[] encoded = new byte[originalData.length + offset]; - System.arraycopy(compressedData, 0, encoded, 0, offset); - System.arraycopy(originalData, 0, encoded, offset, originalData.length); return encoded; } - @Nonnull - private static byte[] decompressIfNeeded(boolean decompressionNeeded, @Nonnull byte[] data, int offset) { - if (!decompressionNeeded) { - // Return the array to be the final output for decoding, so the offset bytes need to be removed - return Arrays.copyOfRange(data, offset, data.length); + private void decompressIfNeeded(@Nonnull EncodingState state, @Nonnull ByteArrayDataInput encodedDataInput) + throws IOException { + if (!state.isCompressed()) { + return; } - if (data.length < 2 + offset) { - throw new RecordCoreException("Invalid data for decompression") - .addLogInfo(LogMessageKeys.DIR_VALUE, data); - } - byte version = data[offset]; + final byte version = encodedDataInput.readByte(); if (version != COMPRESSION_VERSION_FOR_HIGH_COMPRESSION) { throw new RecordCoreException("Un-supported compression version") .addLogInfo(LuceneLogMessageKeys.COMPRESSION_VERSION, version); } + BytesRef ref = new BytesRef(); + int originalLength = encodedDataInput.readVInt(); + Decompressor decompressor = CompressionMode.HIGH_COMPRESSION.newDecompressor(); + decompressor.decompress(encodedDataInput, originalLength, 0, originalLength, ref); + encodedDataInput.reset(ref.bytes, ref.offset, ref.length); + } + + private byte[] encryptIfNeeded(@Nonnull EncodingState state, @Nonnull byte[] encoded, int prefixLength) throws GeneralSecurityException { + if (!state.isEncrypted()) { + return encoded; + } + + final byte[] encrypted; + final byte[] ivData = new byte[CipherPool.IV_SIZE]; + keyManager.getRandom(state.getKeyNumber()).nextBytes(ivData); + final IvParameterSpec iv = new IvParameterSpec(ivData); + final Cipher cipher = CipherPool.borrowCipher(keyManager.getCipher(state.getKeyNumber())); + try { + cipher.init(Cipher.ENCRYPT_MODE, keyManager.getKey(state.getKeyNumber()), iv); + encrypted = cipher.doFinal(encoded, prefixLength, encoded.length - prefixLength); + } finally { + CipherPool.returnCipher(cipher); + } + final int totalSize = prefixLength + CipherPool.IV_SIZE + encrypted.length; + final byte[] withIv = new byte[totalSize]; + System.arraycopy(encoded, 0, withIv, 0, prefixLength); + System.arraycopy(iv.getIV(), 0, withIv, prefixLength, CipherPool.IV_SIZE); + System.arraycopy(encrypted, 0, withIv, prefixLength + CipherPool.IV_SIZE, encrypted.length); + return withIv; + } + + private void decryptIfNeeded(@Nonnull EncodingState state, @Nonnull ByteArrayDataInput encodedDataInput) + throws GeneralSecurityException { + if (!state.isEncrypted()) { + return; + } + + if (keyManager == null) { + throw new RecordCoreException("cannot decrypt Lucene blocks without keys"); + } + + final byte[] ivData = new byte[CipherPool.IV_SIZE]; + encodedDataInput.readBytes(ivData, 0, CipherPool.IV_SIZE); + final IvParameterSpec iv = new IvParameterSpec(ivData); + + final byte[] decrypted; + final byte[] encrypted = new byte[encodedDataInput.length() - encodedDataInput.getPosition()]; + encodedDataInput.readBytes(encrypted, 0, encrypted.length); + final Cipher cipher = CipherPool.borrowCipher(keyManager.getCipher(state.getKeyNumber())); try { - BytesRef ref = new BytesRef(); - ByteArrayDataInput input = new ByteArrayDataInput(data, offset + 1, data.length - offset - 1); - int originalLength = input.readVInt(); - Decompressor decompressor = CompressionMode.HIGH_COMPRESSION.newDecompressor(); - decompressor.decompress(input, originalLength, 0, originalLength, ref); - return Arrays.copyOfRange(ref.bytes, ref.offset, ref.length); - } catch (Exception e) { - throw new RecordCoreException("Lucene data decompression failure", e); + cipher.init(Cipher.DECRYPT_MODE, keyManager.getKey(state.getKeyNumber()), iv); + decrypted = cipher.doFinal(encrypted); + } finally { + CipherPool.returnCipher(cipher); } + encodedDataInput.reset(decrypted); } } diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/FDBLuceneQueryTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/FDBLuceneQueryTest.java index b5566bd081..53df11b3a3 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/FDBLuceneQueryTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/FDBLuceneQueryTest.java @@ -36,6 +36,8 @@ import com.apple.foundationdb.record.metadata.IndexTypes; import com.apple.foundationdb.record.metadata.expressions.GroupingKeyExpression; import com.apple.foundationdb.record.metadata.expressions.KeyExpression; +import com.apple.foundationdb.record.provider.common.RollingTestKeyManager; +import com.apple.foundationdb.record.provider.common.SerializationKeyManager; import com.apple.foundationdb.record.provider.common.text.AllSuffixesTextTokenizer; import com.apple.foundationdb.record.provider.common.text.TextSamples; import com.apple.foundationdb.record.provider.foundationdb.FDBDatabaseFactory; @@ -54,6 +56,7 @@ import com.apple.foundationdb.record.util.pair.Pair; import com.apple.foundationdb.tuple.Tuple; import com.apple.test.BooleanSource; +import com.apple.test.RandomSeedSource; import com.apple.test.RandomizedTestUtils; import com.apple.test.SuperSlow; import com.apple.test.Tags; @@ -208,6 +211,7 @@ public static void setup() { private static final Index COMPLEX_TEXT_BY_GROUP = new Index("Complex$text_by_group", function(LuceneFunctionNames.LUCENE_TEXT, field("text")).groupBy(field("group")), LuceneIndexTypes.LUCENE); private ExecutorService executorService = null; + private SerializationKeyManager keyManager; @Override public void setupPlanner(@Nullable PlannableIndexTypes indexTypes) { @@ -229,7 +233,9 @@ public void setupPlanner(@Nullable PlannableIndexTypes indexTypes) { @Override protected RecordLayerPropertyStorage.Builder addDefaultProps(final RecordLayerPropertyStorage.Builder props) { return super.addDefaultProps(props) - .addProp(LuceneRecordContextProperties.LUCENE_EXECUTOR_SERVICE, (Supplier)() -> executorService); + .addProp(LuceneRecordContextProperties.LUCENE_EXECUTOR_SERVICE, (Supplier)() -> executorService) + .addProp(LuceneRecordContextProperties.LUCENE_INDEX_ENCRYPTION_ENABLED, (Supplier)() -> keyManager != null) + .addProp(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER, (Supplier)() -> keyManager); } @SuppressWarnings("SameParameterValue") @@ -1398,4 +1404,35 @@ void testQueryWithManyDocuments() { assertPrimaryKeys("text:morningstart", false, Set.of()); } } + + @ParameterizedTest + @RandomSeedSource + void encrypted(long seed) throws Exception { + final RollingTestKeyManager rollingTestKeyManager = new RollingTestKeyManager(seed); + keyManager = rollingTestKeyManager; + // Save in many transactions to get more segments and so more blocks. + for (TestRecordsTextProto.SimpleDocument doc : DOCUMENTS) { + try (FDBRecordContext context = openContext()) { + openRecordStore(context); + recordStore.saveRecord(doc); + commit(context); + } + } + assertThat(rollingTestKeyManager.numberOfKeys(), greaterThan(10)); + try (FDBRecordContext context = openContext()) { + openRecordStore(context); + final QueryComponent filter = new LuceneQueryComponent("parents", Lists.newArrayList("text"), true); + RecordQuery.Builder query = RecordQuery.newBuilder() + .setRecordType(TextIndexTestUtils.SIMPLE_DOC) + .setFilter(filter); + RecordQueryPlan plan = planQuery(query.build()); + List primaryKeys; + try (RecordCursor> recordCursor = recordStore.executeQuery(plan)) { + primaryKeys = recordCursor.map(FDBQueriedRecord::getPrimaryKey).map(t -> t.getLong(0)).asList().get(); + } + assertEquals(Set.of(2L, 4L, 5L), new HashSet<>(primaryKeys)); + } + } + + } diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneCompressionTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java similarity index 66% rename from fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneCompressionTest.java rename to fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java index 54b76f78ae..08514b740e 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneCompressionTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java @@ -21,23 +21,30 @@ package com.apple.foundationdb.record.lucene; import com.apple.foundationdb.record.lucene.directory.LuceneSerializer; +import com.apple.foundationdb.record.provider.common.FixedZeroKeyManager; +import com.apple.foundationdb.record.provider.common.SerializationKeyManager; import com.apple.foundationdb.record.util.RandomUtil; +import com.apple.test.BooleanSource; import com.apple.test.Tags; import com.google.protobuf.ByteString; import com.google.protobuf.InvalidProtocolBufferException; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import javax.crypto.KeyGenerator; +import javax.crypto.SecretKey; import java.util.concurrent.ThreadLocalRandom; /** - * Test for Lucene data compression/decompression validation. + * Test for Lucene data compression/decompression and encryption/decryption validation. */ @Tag(Tags.RequiresFDB) -public class LuceneCompressionTest { +class LuceneSerializerTest { @Test void testEncodingWithoutCompression() throws InvalidProtocolBufferException { + final LuceneSerializer serializer = new LuceneSerializer(null); final ByteString content = RandomUtil.randomByteString(ThreadLocalRandom.current(), 100); final LuceneFileSystemProto.LuceneFileReference reference = LuceneFileSystemProto.LuceneFileReference.newBuilder() .setId(1) @@ -46,8 +53,8 @@ void testEncodingWithoutCompression() throws InvalidProtocolBufferException { .setContent(content) .build(); final byte[] originalValue = reference.toByteArray(); - final byte[] encodedValue = LuceneSerializer.encode(originalValue, true, false); - final byte[] decodedValue = LuceneSerializer.decode(encodedValue); + final byte[] encodedValue = serializer.encode(originalValue, true, false); + final byte[] decodedValue = serializer.decode(encodedValue); final byte[] expectedEncodedValue = new byte[originalValue.length + 1]; System.arraycopy(originalValue, 0, expectedEncodedValue, 1, originalValue.length); @@ -67,6 +74,7 @@ void testEncodingWithoutCompression() throws InvalidProtocolBufferException { @Test void testEncodingWithCompression() throws InvalidProtocolBufferException { + final LuceneSerializer serializer = new LuceneSerializer(null); final String duplicateMsg = "abcdefghijklmnopqrstuvwxyz"; final String content = "content_" + duplicateMsg + "_" + duplicateMsg; final LuceneFileSystemProto.LuceneFileReference reference = LuceneFileSystemProto.LuceneFileReference.newBuilder() @@ -76,8 +84,8 @@ void testEncodingWithCompression() throws InvalidProtocolBufferException { .setContent(ByteString.copyFromUtf8(content)) .build(); final byte[] value = reference.toByteArray(); - final byte[] encodedValue = LuceneSerializer.encode(value, true, false); - final byte[] decodedValue = LuceneSerializer.decode(encodedValue); + final byte[] encodedValue = serializer.encode(value, true, false); + final byte[] decodedValue = serializer.decode(encodedValue); // The encoded value's size is smaller than the original one due to compression Assertions.assertTrue(value.length > encodedValue.length); @@ -91,4 +99,27 @@ void testEncodingWithCompression() throws InvalidProtocolBufferException { Assertions.assertEquals(10L, decompressedReference.getBlockSize()); Assertions.assertEquals(ByteString.copyFromUtf8(content), decompressedReference.getContent()); } + + @ParameterizedTest + @BooleanSource + void testEncodingWithEncryption(boolean compressToo) throws Exception { + KeyGenerator keyGen = KeyGenerator.getInstance("AES"); + keyGen.init(128); + SecretKey key = keyGen.generateKey(); + final SerializationKeyManager keyManager = new FixedZeroKeyManager(key, null, null); + final LuceneSerializer serializer = new LuceneSerializer(keyManager); + final ByteString content = RandomUtil.randomByteString(ThreadLocalRandom.current(), 100); + final LuceneFileSystemProto.LuceneFileReference reference = LuceneFileSystemProto.LuceneFileReference.newBuilder() + .setId(1) + .setSize(20L) + .setBlockSize(10L) + .setContent(content) + .build(); + final byte[] value = reference.toByteArray(); + final byte[] encodedValue = serializer.encode(value, compressToo, true); + final byte[] decodedValue = serializer.decode(encodedValue); + Assertions.assertArrayEquals(value, decodedValue); + final LuceneFileSystemProto.LuceneFileReference decryptedReference = LuceneFileSystemProto.LuceneFileReference.parseFrom(decodedValue); + Assertions.assertEquals(content, decryptedReference.getContent()); + } } diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java index 79574c2b94..8e455515d1 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java @@ -121,7 +121,7 @@ public void testWriteSeekData() throws Exception { assertTrue(ex.getCause() instanceof TimeoutException); directory.getCallerContext().commit(); - assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, LuceneSerializer.encode(data, true, false).length); + assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, directory.getSerializer().encode(data, true, false).length); } @Test diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java index 827bd4a676..98364c4f99 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java @@ -129,8 +129,9 @@ public void testWriteLuceneFileReference() { FDBLuceneFileReference luceneFileReference = directory.getFDBLuceneFileReference("test1"); assertNotNull(luceneFileReference, "fileReference should exist"); + LuceneSerializer serializer = directory.getSerializer(); assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE_FILE_REFERENCE, 2, - LuceneSerializer.encode(reference1.getBytes(), true, false).length + LuceneSerializer.encode(reference2.getBytes(), true, false).length); + serializer.encode(reference1.getBytes(), true, false).length + serializer.encode(reference2.getBytes(), true, false).length); } @Test @@ -159,7 +160,7 @@ public void testWriteSeekData() throws Exception { directory.getFDBLuceneFileReferenceAsync("testReference2"), 1).get(), "seek data should exist"); directory.getCallerContext().commit(); - assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, LuceneSerializer.encode(data, true, false).length); + assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, directory.getSerializer().encode(data, true, false).length); } @Test From 15428913a7b14017b537576f47436d1b41c0373d Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Fri, 26 Sep 2025 07:02:52 -0700 Subject: [PATCH 2/9] Extract a common subclass --- ...CompressedAndEncryptedSerializerState.java | 61 +++++++++++++++++++ .../TransformedRecordSerializerState.java | 31 +--------- .../lucene/directory/LuceneSerializer.java | 51 +++------------- 3 files changed, 70 insertions(+), 73 deletions(-) create mode 100644 fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java new file mode 100644 index 0000000000..4670b66d52 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java @@ -0,0 +1,61 @@ +/* + * CompressedAndEncryptedSerializerState.java + * + * This source file is part of the FoundationDB open source project + * + * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors + * + * Licensed 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 com.apple.foundationdb.record.provider.common; + +import com.apple.foundationdb.annotation.API; + +/** + * Information on intended / found serialization format: compressed and/or encrypted. + */ +public class CompressedAndEncryptedSerializerState { + private boolean compressed; + private boolean encrypted; + private int keyNumber; + + public CompressedAndEncryptedSerializerState() { + this.compressed = false; + this.encrypted = false; + } + + public boolean isCompressed() { + return compressed; + } + + public void setCompressed(boolean compressed) { + this.compressed = compressed; + } + + public boolean isEncrypted() { + return encrypted; + } + + public void setEncrypted(boolean encrypted) { + this.encrypted = encrypted; + } + + public int getKeyNumber() { + return keyNumber; + } + + public void setKeyNumber(final int keyNumber) { + this.keyNumber = keyNumber; + } +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/TransformedRecordSerializerState.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/TransformedRecordSerializerState.java index 6cdcb90f9b..34842b9c61 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/TransformedRecordSerializerState.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/TransformedRecordSerializerState.java @@ -28,15 +28,10 @@ /** * The internal state of serialization / deserialization, pointing to a portion of a byte array. - * Also includes information on intended / found serialization format. */ @API(API.Status.INTERNAL) @SpotBugsSuppressWarnings("EI_EXPOSE_REP") -class TransformedRecordSerializerState { - private boolean compressed; - private boolean encrypted; - private int keyNumber; - +class TransformedRecordSerializerState extends CompressedAndEncryptedSerializerState { @Nonnull private byte[] data; private int offset; @@ -52,30 +47,6 @@ public TransformedRecordSerializerState(@Nonnull byte[] data, int offset, int le this.length = length; } - public boolean isCompressed() { - return compressed; - } - - public void setCompressed(boolean compressed) { - this.compressed = compressed; - } - - public boolean isEncrypted() { - return encrypted; - } - - public void setEncrypted(boolean encrypted) { - this.encrypted = encrypted; - } - - public int getKeyNumber() { - return keyNumber; - } - - public void setKeyNumber(int keyNumber) { - this.keyNumber = keyNumber; - } - @Nonnull public byte[] getData() { return data; diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java index 80df13283b..e9c8cda10b 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java @@ -24,6 +24,7 @@ import com.apple.foundationdb.record.logging.KeyValueLogMessage; import com.apple.foundationdb.record.lucene.LuceneLogMessageKeys; import com.apple.foundationdb.record.provider.common.CipherPool; +import com.apple.foundationdb.record.provider.common.CompressedAndEncryptedSerializerState; import com.apple.foundationdb.record.provider.common.SerializationKeyManager; import org.apache.lucene.codecs.compressing.CompressionMode; import org.apache.lucene.codecs.compressing.Compressor; @@ -59,54 +60,18 @@ public LuceneSerializer(@Nullable SerializationKeyManager keyManager) { this.keyManager = keyManager; } - private static class EncodingState { - private boolean compressed; - private boolean encrypted; - private int keyNumber; - - private EncodingState() { - this.compressed = false; - this.encrypted = false; - } - - boolean isCompressed() { - return compressed; - } - - void setCompressed(boolean compressed) { - this.compressed = compressed; - } - - boolean isEncrypted() { - return encrypted; - } - - void setEncrypted(boolean encrypted) { - this.encrypted = encrypted; - } - - public int getKeyNumber() { - return keyNumber; - } - - public void setKeyNumber(final int keyNumber) { - this.keyNumber = keyNumber; - } - } - @Nullable public byte[] encode(@Nullable byte[] data, boolean compress, boolean encrypt) { if (data == null) { return null; } - final EncodingState state = new EncodingState(); + final CompressedAndEncryptedSerializerState state = new CompressedAndEncryptedSerializerState(); long prefix = 0; if (compress) { prefix |= ENCODING_COMPRESSED; state.setCompressed(true); } - // Encryption will be supported in future if (encrypt) { if (keyManager == null) { throw new RecordCoreException("cannot encrypt Lucene blocks without keys"); @@ -153,7 +118,7 @@ public byte[] decode(@Nullable byte[] data) { .addLogInfo(LuceneLogMessageKeys.DATA_VALUE, data); } - final EncodingState state = new EncodingState(); + final CompressedAndEncryptedSerializerState state = new CompressedAndEncryptedSerializerState(); byte[] decoded; try { final ByteArrayDataInput encodedDataInput = new ByteArrayDataInput(data); @@ -184,7 +149,7 @@ public byte[] decode(@Nullable byte[] data) { } @Nonnull - private static byte[] compressIfNeeded(@Nonnull EncodingState state, @Nonnull ByteBuffersDataOutput encodedDataOutput, + private static byte[] compressIfNeeded(@Nonnull CompressedAndEncryptedSerializerState state, @Nonnull ByteBuffersDataOutput encodedDataOutput, @Nonnull byte[] uncompressedData, int prefixLength) throws IOException { if (!state.isCompressed()) { @@ -207,7 +172,7 @@ private static byte[] compressIfNeeded(@Nonnull EncodingState state, @Nonnull By } } - private static byte[] fallBackToUncompressed(@Nonnull EncodingState state, @Nonnull byte[] originalData, + private static byte[] fallBackToUncompressed(@Nonnull CompressedAndEncryptedSerializerState state, @Nonnull byte[] originalData, @Nonnull byte[] encodedData, int prefixLength) { final byte[] encoded = new byte[originalData.length + prefixLength]; System.arraycopy(encodedData, 0, encoded, 0, prefixLength); @@ -217,7 +182,7 @@ private static byte[] fallBackToUncompressed(@Nonnull EncodingState state, @Nonn return encoded; } - private void decompressIfNeeded(@Nonnull EncodingState state, @Nonnull ByteArrayDataInput encodedDataInput) + private void decompressIfNeeded(@Nonnull CompressedAndEncryptedSerializerState state, @Nonnull ByteArrayDataInput encodedDataInput) throws IOException { if (!state.isCompressed()) { return; @@ -236,7 +201,7 @@ private void decompressIfNeeded(@Nonnull EncodingState state, @Nonnull ByteArray encodedDataInput.reset(ref.bytes, ref.offset, ref.length); } - private byte[] encryptIfNeeded(@Nonnull EncodingState state, @Nonnull byte[] encoded, int prefixLength) throws GeneralSecurityException { + private byte[] encryptIfNeeded(@Nonnull CompressedAndEncryptedSerializerState state, @Nonnull byte[] encoded, int prefixLength) throws GeneralSecurityException { if (!state.isEncrypted()) { return encoded; } @@ -260,7 +225,7 @@ private byte[] encryptIfNeeded(@Nonnull EncodingState state, @Nonnull byte[] enc return withIv; } - private void decryptIfNeeded(@Nonnull EncodingState state, @Nonnull ByteArrayDataInput encodedDataInput) + private void decryptIfNeeded(@Nonnull CompressedAndEncryptedSerializerState state, @Nonnull ByteArrayDataInput encodedDataInput) throws GeneralSecurityException { if (!state.isEncrypted()) { return; From 2a39e0d767353a4187900e87dec1be8938a3527d Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Fri, 26 Sep 2025 07:05:06 -0700 Subject: [PATCH 3/9] Add clarifying comment --- .../foundationdb/record/lucene/directory/LuceneSerializer.java | 1 + 1 file changed, 1 insertion(+) diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java index e9c8cda10b..2aa0491ee8 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java @@ -176,6 +176,7 @@ private static byte[] fallBackToUncompressed(@Nonnull CompressedAndEncryptedSeri @Nonnull byte[] encodedData, int prefixLength) { final byte[] encoded = new byte[originalData.length + prefixLength]; System.arraycopy(encodedData, 0, encoded, 0, prefixLength); + // This bit is always in the lowest (first) byte, even if the prefix is longer. encoded[0] &= ~ENCODING_COMPRESSED; System.arraycopy(originalData, 0, encoded, prefixLength, originalData.length); state.setCompressed(false); From 6c93a24943ac6eaed9ba7688592f49be75d25433 Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Fri, 26 Sep 2025 07:45:37 -0700 Subject: [PATCH 4/9] Randomize compression and encryption in LuceneIndexMaintenanceTest --- .../RecordLayerPropertyStorage.java | 4 ++ .../lucene/LuceneIndexMaintenanceTest.java | 56 ++++++++++++------- 2 files changed, 41 insertions(+), 19 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java index 612ec6b1b3..32f55a5905 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java @@ -96,6 +96,10 @@ private Builder(ImmutableMap, RecordLayerPropertyValue this.propertyMap = new HashMap<>(properties); } + public boolean hasProp(@Nonnull RecordLayerPropertyKey propKey) { + return propertyMap.containsKey(propKey); + } + public Builder addProp(@Nonnull RecordLayerPropertyValue propValue) { if (this.propertyMap.putIfAbsent(propValue.getKey(), propValue) != null) { throw new RecordCoreException("Duplicate property name is added") diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java index 8ba8f1cfe0..160c15454e 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java @@ -35,6 +35,7 @@ import com.apple.foundationdb.record.lucene.directory.FDBDirectoryLockFactory; import com.apple.foundationdb.record.lucene.directory.FDBDirectoryWrapper; import com.apple.foundationdb.record.metadata.Index; +import com.apple.foundationdb.record.provider.common.RollingTestKeyManager; import com.apple.foundationdb.record.provider.common.StoreTimer; import com.apple.foundationdb.record.provider.foundationdb.FDBExceptions; import com.apple.foundationdb.record.provider.foundationdb.FDBRecordContext; @@ -70,6 +71,7 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; import java.io.IOException; +import java.security.GeneralSecurityException; import java.time.Duration; import java.util.ArrayDeque; import java.util.HashMap; @@ -250,11 +252,13 @@ void savingInReverseDoesNotRequireRepartitioning(boolean isGrouped, static Stream manyDocumentsArgumentsSlow() { return Stream.concat( - Stream.of(Arguments.of(true, true, true, 80, 2, 200, 234809), + Stream.of(Arguments.of(true, true, true, true, false, 80, 2, 200, 234809), // I don't know why, but this took over an hour, I'm hoping my laptop slept, but I don't see it - Arguments.of(false, true, false, 50, 8, 212, 3125111852333110588L)), + Arguments.of(false, true, false, true, false, 50, 8, 212, 3125111852333110588L)), RandomizedTestUtils.randomArguments(random -> Arguments.of(random.nextBoolean(), + random.nextBoolean(), + random.nextBoolean(), random.nextBoolean(), random.nextBoolean(), // We want to have a high partitionHighWatermark so that the underlying lucene indexes @@ -271,11 +275,13 @@ static Stream manyDocumentsArgumentsSlow() { void manyDocumentSlow(boolean isGrouped, boolean isSynthetic, boolean primaryKeySegmentIndexEnabled, + boolean compressed, + boolean encrypted, int partitionHighWatermark, int repartitionCount, int loopCount, - long seed) throws IOException { - manyDocument(isGrouped, isSynthetic, primaryKeySegmentIndexEnabled, partitionHighWatermark, + long seed) throws IOException, GeneralSecurityException { + manyDocuments(isGrouped, isSynthetic, primaryKeySegmentIndexEnabled, compressed, encrypted, partitionHighWatermark, repartitionCount, loopCount, 10, seed); } @@ -283,15 +289,17 @@ void manyDocumentSlow(boolean isGrouped, static Stream manyDocumentsArguments() { return Stream.concat( Stream.concat( - Stream.of(Arguments.of(true, true, true, 20, 4, 50, 3, -644766138635622644L)), + Stream.of(Arguments.of(true, true, true, true, false, 20, 4, 50, 3, -644766138635622644L)), TestConfigurationUtils.onlyNightly( Stream.of( - Arguments.of(true, false, false, 21, 3, 55, 3, 9237590782644L), - Arguments.of(false, true, true, 18, 3, 46, 3, -1089113174774589435L), - Arguments.of(false, false, false, 24, 6, 59, 3, 6223372946177329440L), - Arguments.of(true, false, false, 27, 9, 48, 3, 2451719304283565963L)))), + Arguments.of(true, false, false, true, false, 21, 3, 55, 3, 9237590782644L), + Arguments.of(false, true, true, true, false, 18, 3, 46, 3, -1089113174774589435L), + Arguments.of(false, false, false, true, false, 24, 6, 59, 3, 6223372946177329440L), + Arguments.of(true, false, false, true, false, 27, 9, 48, 3, 2451719304283565963L)))), RandomizedTestUtils.randomArguments(random -> Arguments.of(random.nextBoolean(), + random.nextBoolean(), + random.nextBoolean(), random.nextBoolean(), random.nextBoolean(), // We want to have a high partitionHighWatermark so that the underlying lucene indexes @@ -305,14 +313,16 @@ static Stream manyDocumentsArguments() { @ParameterizedTest @MethodSource("manyDocumentsArguments") - void manyDocument(boolean isGrouped, - boolean isSynthetic, - boolean primaryKeySegmentIndexEnabled, - int partitionHighWatermark, - int repartitionCount, - int loopCount, - int maxTransactionsPerLoop, - long seed) throws IOException { + void manyDocuments(boolean isGrouped, + boolean isSynthetic, + boolean primaryKeySegmentIndexEnabled, + boolean compressed, + boolean encrypted, + int partitionHighWatermark, + int repartitionCount, + int loopCount, + int maxTransactionsPerLoop, + long seed) throws IOException, GeneralSecurityException { final LuceneIndexTestDataModel dataModel = new LuceneIndexTestDataModel.Builder(seed, this::getStoreBuilder, pathManager) .setIsGrouped(isGrouped) .setIsSynthetic(isSynthetic) @@ -326,11 +336,16 @@ void manyDocument(boolean isGrouped, "seed", seed, "loopCount", loopCount)); - final RecordLayerPropertyStorage contextProps = RecordLayerPropertyStorage.newBuilder() + final RecordLayerPropertyStorage.Builder contextPropsBuilder = RecordLayerPropertyStorage.newBuilder() .addProp(LuceneRecordContextProperties.LUCENE_REPARTITION_DOCUMENT_COUNT, repartitionCount) .addProp(LuceneRecordContextProperties.LUCENE_MAX_DOCUMENTS_TO_MOVE_DURING_REPARTITIONING, dataModel.nextInt(1000) + repartitionCount) .addProp(LuceneRecordContextProperties.LUCENE_MERGE_SEGMENTS_PER_TIER, (double)dataModel.nextInt(10) + 2) // it must be at least 2.0 - .build(); + .addProp(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED, compressed) + .addProp(LuceneRecordContextProperties.LUCENE_INDEX_ENCRYPTION_ENABLED, encrypted); + if (encrypted) { + contextPropsBuilder.addProp(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER, new RollingTestKeyManager(seed)); + } + final RecordLayerPropertyStorage contextProps = contextPropsBuilder.build(); for (int i = 0; i < loopCount; i++) { LOGGER.info(KeyValueLogMessage.of("ManyDocument loop", "iteration", i, @@ -1279,6 +1294,9 @@ private void explicitMergeIndex(Index index, } protected RecordLayerPropertyStorage.Builder addDefaultProps(final RecordLayerPropertyStorage.Builder props) { + if (props.hasProp(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED)) { + return props; + } return super.addDefaultProps(props).addProp(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED, true); } } From 0f58b0e403f770bbc9be43e869c080eb2dfb5200 Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Fri, 26 Sep 2025 09:03:17 -0700 Subject: [PATCH 5/9] Add a metric for the number of Lucene block writes --- .../java/com/apple/foundationdb/record/lucene/LuceneEvents.java | 2 ++ .../foundationdb/record/lucene/directory/FDBDirectory.java | 1 + 2 files changed, 3 insertions(+) diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneEvents.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneEvents.java index 30347ecee1..e8a6b11089 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneEvents.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneEvents.java @@ -209,6 +209,8 @@ public enum Counts implements StoreTimer.Count { LUCENE_GET_INCREMENT_CALLS("lucene increments", false), /** The number of block reads that occur against the FDBDirectory.*/ LUCENE_BLOCK_READS("lucene block reads", false), + /** The number of block writes that occur against the FDBDirectory.*/ + LUCENE_BLOCK_WRITES("lucene block writes", false), /** Matched documents returned from lucene index reader scans. **/ LUCENE_SCAN_MATCHED_DOCUMENTS("lucene scan matched documents", false), /** Matched auto complete suggestions returned from lucene auto complete suggestion lookup. **/ diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java index b5971b0f89..4fb81ce187 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java @@ -428,6 +428,7 @@ public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLucene */ public int writeData(final long id, final int block, @Nonnull final byte[] value) { final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(value, compressionEnabled, encryptionEnabled)); + agilityContext.increment(LuceneEvents.Counts.LUCENE_BLOCK_WRITES); //This may not be correct transactionally agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE, encodedBytes.length); if (LOGGER.isTraceEnabled()) { From 127e8cde82facad2e7a2a7d9c46a00805c74da08 Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Mon, 29 Sep 2025 08:59:26 -0700 Subject: [PATCH 6/9] Update fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java Co-authored-by: ohadzeliger <70664918+ohadzeliger@users.noreply.github.com> --- .../foundationdb/record/lucene/directory/LuceneSerializer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java index 2aa0491ee8..553910fb1d 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java @@ -140,7 +140,7 @@ public byte[] decode(@Nullable byte[] data) { if (LOGGER.isTraceEnabled()) { LOGGER.trace(KeyValueLogMessage.of("Decoded lucene data", LuceneLogMessageKeys.COMPRESSED_EVENTUALLY, state.isCompressed(), - LuceneLogMessageKeys.ENCRYPTED_EVENTUALLY, state.isCompressed(), + LuceneLogMessageKeys.ENCRYPTED_EVENTUALLY, state.isEncrypted(), LuceneLogMessageKeys.ENCODED_DATA_SIZE, data.length, LuceneLogMessageKeys.ORIGINAL_DATA_SIZE, decoded.length)); } From 8c71285ac3b8436943a42709a6b3ec2685677ecd Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Mon, 29 Sep 2025 10:01:57 -0700 Subject: [PATCH 7/9] Add a test of failing decryption in Lucene --- .../RecordLayerPropertyStorage.java | 4 ++ .../lucene/LuceneIndexMaintenanceTest.java | 52 +++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java index 32f55a5905..c52c65d7d2 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java @@ -100,6 +100,10 @@ public boolean hasProp(@Nonnull RecordLayerPropertyKey propKey) { return propertyMap.containsKey(propKey); } + public void removeProp(@Nonnull RecordLayerPropertyKey propKey) { + propertyMap.remove(propKey); + } + public Builder addProp(@Nonnull RecordLayerPropertyValue propValue) { if (this.propertyMap.putIfAbsent(propValue.getKey(), propValue) != null) { throw new RecordCoreException("Duplicate property name is added") diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java index 160c15454e..22af6e6ccb 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintenanceTest.java @@ -35,7 +35,9 @@ import com.apple.foundationdb.record.lucene.directory.FDBDirectoryLockFactory; import com.apple.foundationdb.record.lucene.directory.FDBDirectoryWrapper; import com.apple.foundationdb.record.metadata.Index; +import com.apple.foundationdb.record.provider.common.FixedZeroKeyManager; import com.apple.foundationdb.record.provider.common.RollingTestKeyManager; +import com.apple.foundationdb.record.provider.common.SerializationKeyManager; import com.apple.foundationdb.record.provider.common.StoreTimer; import com.apple.foundationdb.record.provider.foundationdb.FDBExceptions; import com.apple.foundationdb.record.provider.foundationdb.FDBRecordContext; @@ -70,6 +72,8 @@ import javax.annotation.Nonnull; import javax.annotation.Nullable; +import javax.crypto.KeyGenerator; +import javax.crypto.SecretKey; import java.io.IOException; import java.security.GeneralSecurityException; import java.time.Duration; @@ -104,6 +108,8 @@ import static com.apple.foundationdb.record.metadata.Key.Expressions.field; import static com.apple.foundationdb.record.metadata.Key.Expressions.function; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -893,6 +899,52 @@ void sampledDelete(boolean isSynthetic, boolean isGrouped, long seed) throws IOE assertThat(partitionCounts, Matchers.contains(5, 3, 4))); } + static Stream changingEncryptionKey() { + return Stream.concat(Stream.of(Arguments.of(true, true, 288513), + Arguments.of(false, false, 792025)), + RandomizedTestUtils.randomArguments(random -> + Arguments.of(random.nextBoolean(), random.nextBoolean(), random.nextLong()))); + } + + @ParameterizedTest + @MethodSource + void changingEncryptionKey(boolean isSynthetic, boolean isGrouped, long seed) throws IOException, GeneralSecurityException { + final LuceneIndexTestDataModel dataModel = new LuceneIndexTestDataModel.Builder(seed, this::getStoreBuilder, pathManager) + .setIsGrouped(isGrouped) + .setIsSynthetic(isSynthetic) + .setPrimaryKeySegmentIndexEnabled(true) + .build(); + + final RecordLayerPropertyStorage.Builder contextPropsBuilder = RecordLayerPropertyStorage.newBuilder() + .addProp(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED, true) + .addProp(LuceneRecordContextProperties.LUCENE_INDEX_ENCRYPTION_ENABLED, true); + final KeyGenerator keyGen = KeyGenerator.getInstance("AES"); + keyGen.init(128); + final SecretKey key1 = keyGen.generateKey(); + final SerializationKeyManager keyManager1 = new FixedZeroKeyManager(key1, null, null); + contextPropsBuilder.addProp(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER, keyManager1); + final RecordLayerPropertyStorage contextProps1 = contextPropsBuilder.build(); + + try (FDBRecordContext context = openContext(contextProps1)) { + dataModel.saveRecordsToAllGroups(20, context); + commit(context); + } + + explicitMergeIndex(dataModel.index, contextProps1, dataModel.schemaSetup); + + final SecretKey key2 = keyGen.generateKey(); + final SerializationKeyManager keyManager2 = new FixedZeroKeyManager(key2, null, null); + contextPropsBuilder.removeProp(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER); + contextPropsBuilder.addProp(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER, keyManager2); + final RecordLayerPropertyStorage contextProps2 = contextPropsBuilder.build(); + IOException ioException = assertThrows(IOException.class, + () -> dataModel.validate(() -> openContext(contextProps2))); + assertThat(ioException.getCause(), instanceOf(RecordCoreException.class)); + assertThat(ioException.getCause().getMessage(), containsString("Lucene data decoding failure")); + assertThat(ioException.getCause().getCause(), instanceOf(GeneralSecurityException.class)); + + } + private static Stream concurrentParameters() { // only run the individual tests with synthetic during nightly, the mix runs both return Stream.concat(Stream.of(false), From 4749ea1b8fcaed5d630f46eef016e7d76f35a1fc Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Mon, 29 Sep 2025 10:35:54 -0700 Subject: [PATCH 8/9] Move the {compression,encryption}Enabled state gotten from properties from the directory to its serializer. --- .../record/lucene/directory/FDBDirectory.java | 23 +++++++------- .../lucene/directory/LuceneSerializer.java | 30 +++++++++++++++---- .../record/lucene/LuceneSerializerTest.java | 12 ++++---- .../directory/FDBDirectoryFailuresTest.java | 2 +- .../lucene/directory/FDBDirectoryTest.java | 4 +-- 5 files changed, 43 insertions(+), 28 deletions(-) diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java index 4fb81ce187..5cc47ab9f6 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectory.java @@ -166,9 +166,6 @@ public class FDBDirectory extends Directory { private final Cache, CompletableFuture> blockCache; - private final boolean compressionEnabled; - private final boolean encryptionEnabled; - // The shared cache is initialized when first listing the directory, if a manager is present, and cleared before writing. @Nullable private final FDBDirectorySharedCacheManager sharedCacheManager; @@ -230,9 +227,9 @@ private FDBDirectory(@Nonnull Subspace subspace, @Nullable Map i .removalListener(notification -> cacheRemovalCallback()) .build(); this.fileSequenceCounter = new AtomicLong(-1); - this.compressionEnabled = Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED), false); - this.encryptionEnabled = Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_ENCRYPTION_ENABLED), false); - this.serializer = new LuceneSerializer(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER)); + this.serializer = new LuceneSerializer(Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_COMPRESSION_ENABLED), false), + Objects.requireNonNullElse(agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_ENCRYPTION_ENABLED), false), + agilityContext.getPropertyValue(LuceneRecordContextProperties.LUCENE_INDEX_KEY_MANAGER)); this.fileReferenceMapSupplier = Suppliers.memoize(this::loadFileReferenceCacheForMemoization); this.sharedCacheManager = sharedCacheManager; this.sharedCacheKey = sharedCacheKey; @@ -405,7 +402,7 @@ public static boolean isStoredFieldsFile(String name) { */ public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLuceneFileReference reference) { final byte[] fileReferenceBytes = reference.getBytes(); - final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(fileReferenceBytes, compressionEnabled, encryptionEnabled)); + final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(fileReferenceBytes)); agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE_FILE_REFERENCE, encodedBytes.length); if (LOGGER.isTraceEnabled()) { LOGGER.trace(getLogMessage("Write lucene file reference", @@ -427,7 +424,7 @@ public void writeFDBLuceneFileReference(@Nonnull String name, @Nonnull FDBLucene * @return the actual data size written to database with potential compression and encryption applied */ public int writeData(final long id, final int block, @Nonnull final byte[] value) { - final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(value, compressionEnabled, encryptionEnabled)); + final byte[] encodedBytes = Objects.requireNonNull(serializer.encode(value)); agilityContext.increment(LuceneEvents.Counts.LUCENE_BLOCK_WRITES); //This may not be correct transactionally agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE, encodedBytes.length); @@ -915,10 +912,10 @@ public void rename(@Nonnull final String source, @Nonnull final String dest) thr .addLogInfo(LogMessageKeys.SOURCE_FILE, source) .addLogInfo(LogMessageKeys.INDEX_TYPE, LuceneIndexTypes.LUCENE) .addLogInfo(LogMessageKeys.SUBSPACE, subspace) - .addLogInfo(LuceneLogMessageKeys.COMPRESSION_SUPPOSED, compressionEnabled) - .addLogInfo(LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, encryptionEnabled); + .addLogInfo(LuceneLogMessageKeys.COMPRESSION_SUPPOSED, serializer.isCompressionEnabled()) + .addLogInfo(LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, serializer.isEncryptionEnabled()); } - byte[] encodedBytes = serializer.encode(value.getBytes(), compressionEnabled, encryptionEnabled); + byte[] encodedBytes = serializer.encode(value.getBytes()); agilityContext.set(metaSubspace.pack(dest), encodedBytes); agilityContext.clear(key); @@ -1057,8 +1054,8 @@ private String getLogMessage(@Nonnull String staticMsg, @Nullable final Object.. private KeyValueLogMessage getKeyValueLogMessage(final @Nonnull String staticMsg, final Object... keysAndValues) { return KeyValueLogMessage.build(staticMsg, keysAndValues) .addKeyAndValue(LogMessageKeys.SUBSPACE, subspace) - .addKeyAndValue(LuceneLogMessageKeys.COMPRESSION_SUPPOSED, compressionEnabled) - .addKeyAndValue(LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, encryptionEnabled); + .addKeyAndValue(LuceneLogMessageKeys.COMPRESSION_SUPPOSED, serializer.isCompressionEnabled()) + .addKeyAndValue(LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, serializer.isEncryptionEnabled()); } /** diff --git a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java index 553910fb1d..c1d589d167 100644 --- a/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java +++ b/fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/LuceneSerializer.java @@ -53,26 +53,44 @@ public class LuceneSerializer { private static final byte COMPRESSION_VERSION_FOR_HIGH_COMPRESSION = 0; private static final int ENCRYPTION_KEY_SHIFT = 3; + private final boolean compressionEnabled; + private final boolean encryptionEnabled; @Nullable private final SerializationKeyManager keyManager; - public LuceneSerializer(@Nullable SerializationKeyManager keyManager) { + public LuceneSerializer(boolean compressionEnabled, + boolean encryptionEnabled, @Nullable SerializationKeyManager keyManager) { + this.compressionEnabled = compressionEnabled; + this.encryptionEnabled = encryptionEnabled; this.keyManager = keyManager; } + public boolean isCompressionEnabled() { + return compressionEnabled; + } + + public boolean isEncryptionEnabled() { + return encryptionEnabled; + } + + @Nullable + public SerializationKeyManager getKeyManager() { + return keyManager; + } + @Nullable - public byte[] encode(@Nullable byte[] data, boolean compress, boolean encrypt) { + public byte[] encode(@Nullable byte[] data) { if (data == null) { return null; } final CompressedAndEncryptedSerializerState state = new CompressedAndEncryptedSerializerState(); long prefix = 0; - if (compress) { + if (compressionEnabled) { prefix |= ENCODING_COMPRESSED; state.setCompressed(true); } - if (encrypt) { + if (encryptionEnabled) { if (keyManager == null) { throw new RecordCoreException("cannot encrypt Lucene blocks without keys"); } @@ -96,8 +114,8 @@ public byte[] encode(@Nullable byte[] data, boolean compress, boolean encrypt) { if (LOGGER.isTraceEnabled()) { LOGGER.trace(KeyValueLogMessage.of("Encoded lucene data", - LuceneLogMessageKeys.COMPRESSION_SUPPOSED, compress, - LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, encrypt, + LuceneLogMessageKeys.COMPRESSION_SUPPOSED, compressionEnabled, + LuceneLogMessageKeys.ENCRYPTION_SUPPOSED, encryptionEnabled, LuceneLogMessageKeys.COMPRESSED_EVENTUALLY, state.isCompressed(), LuceneLogMessageKeys.ENCRYPTED_EVENTUALLY, state.isEncrypted(), LuceneLogMessageKeys.ORIGINAL_DATA_SIZE, data.length, diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java index 08514b740e..9ccf947ef3 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/LuceneSerializerTest.java @@ -44,7 +44,7 @@ class LuceneSerializerTest { @Test void testEncodingWithoutCompression() throws InvalidProtocolBufferException { - final LuceneSerializer serializer = new LuceneSerializer(null); + final LuceneSerializer serializer = new LuceneSerializer(true, false, null); final ByteString content = RandomUtil.randomByteString(ThreadLocalRandom.current(), 100); final LuceneFileSystemProto.LuceneFileReference reference = LuceneFileSystemProto.LuceneFileReference.newBuilder() .setId(1) @@ -53,7 +53,7 @@ void testEncodingWithoutCompression() throws InvalidProtocolBufferException { .setContent(content) .build(); final byte[] originalValue = reference.toByteArray(); - final byte[] encodedValue = serializer.encode(originalValue, true, false); + final byte[] encodedValue = serializer.encode(originalValue); final byte[] decodedValue = serializer.decode(encodedValue); final byte[] expectedEncodedValue = new byte[originalValue.length + 1]; @@ -74,7 +74,7 @@ void testEncodingWithoutCompression() throws InvalidProtocolBufferException { @Test void testEncodingWithCompression() throws InvalidProtocolBufferException { - final LuceneSerializer serializer = new LuceneSerializer(null); + final LuceneSerializer serializer = new LuceneSerializer(true, false, null); final String duplicateMsg = "abcdefghijklmnopqrstuvwxyz"; final String content = "content_" + duplicateMsg + "_" + duplicateMsg; final LuceneFileSystemProto.LuceneFileReference reference = LuceneFileSystemProto.LuceneFileReference.newBuilder() @@ -84,7 +84,7 @@ void testEncodingWithCompression() throws InvalidProtocolBufferException { .setContent(ByteString.copyFromUtf8(content)) .build(); final byte[] value = reference.toByteArray(); - final byte[] encodedValue = serializer.encode(value, true, false); + final byte[] encodedValue = serializer.encode(value); final byte[] decodedValue = serializer.decode(encodedValue); // The encoded value's size is smaller than the original one due to compression @@ -107,7 +107,7 @@ void testEncodingWithEncryption(boolean compressToo) throws Exception { keyGen.init(128); SecretKey key = keyGen.generateKey(); final SerializationKeyManager keyManager = new FixedZeroKeyManager(key, null, null); - final LuceneSerializer serializer = new LuceneSerializer(keyManager); + final LuceneSerializer serializer = new LuceneSerializer(compressToo, true, keyManager); final ByteString content = RandomUtil.randomByteString(ThreadLocalRandom.current(), 100); final LuceneFileSystemProto.LuceneFileReference reference = LuceneFileSystemProto.LuceneFileReference.newBuilder() .setId(1) @@ -116,7 +116,7 @@ void testEncodingWithEncryption(boolean compressToo) throws Exception { .setContent(content) .build(); final byte[] value = reference.toByteArray(); - final byte[] encodedValue = serializer.encode(value, compressToo, true); + final byte[] encodedValue = serializer.encode(value); final byte[] decodedValue = serializer.decode(encodedValue); Assertions.assertArrayEquals(value, decodedValue); final LuceneFileSystemProto.LuceneFileReference decryptedReference = LuceneFileSystemProto.LuceneFileReference.parseFrom(decodedValue); diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java index 8e455515d1..4f88f9c263 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryFailuresTest.java @@ -121,7 +121,7 @@ public void testWriteSeekData() throws Exception { assertTrue(ex.getCause() instanceof TimeoutException); directory.getCallerContext().commit(); - assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, directory.getSerializer().encode(data, true, false).length); + assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, directory.getSerializer().encode(data).length); } @Test diff --git a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java index 98364c4f99..a121b6f761 100644 --- a/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java +++ b/fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryTest.java @@ -131,7 +131,7 @@ public void testWriteLuceneFileReference() { LuceneSerializer serializer = directory.getSerializer(); assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE_FILE_REFERENCE, 2, - serializer.encode(reference1.getBytes(), true, false).length + serializer.encode(reference2.getBytes(), true, false).length); + serializer.encode(reference1.getBytes()).length + serializer.encode(reference2.getBytes()).length); } @Test @@ -160,7 +160,7 @@ public void testWriteSeekData() throws Exception { directory.getFDBLuceneFileReferenceAsync("testReference2"), 1).get(), "seek data should exist"); directory.getCallerContext().commit(); - assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, directory.getSerializer().encode(data, true, false).length); + assertCorrectMetricSize(LuceneEvents.SizeEvents.LUCENE_WRITE, 1, directory.getSerializer().encode(data).length); } @Test From c7238601747731a328aa61f1eef56ab32cd771a9 Mon Sep 17 00:00:00 2001 From: Mike McMahon Date: Mon, 29 Sep 2025 10:52:17 -0700 Subject: [PATCH 9/9] Fix missing annotation --- .../provider/common/CompressedAndEncryptedSerializerState.java | 1 + 1 file changed, 1 insertion(+) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java index 4670b66d52..ced339920d 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java @@ -25,6 +25,7 @@ /** * Information on intended / found serialization format: compressed and/or encrypted. */ +@API(API.Status.INTERNAL) public class CompressedAndEncryptedSerializerState { private boolean compressed; private boolean encrypted;