Make Lucene serialization able to encrypt, …

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/common/CompressedAndEncryptedSerializerState.java

@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+@API(API.Status.INTERNAL)
+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;
+    }
+}

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;

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/properties/RecordLayerPropertyStorage.java

@@ -96,6 +96,14 @@ private Builder(ImmutableMap<RecordLayerPropertyKey<?>, RecordLayerPropertyValue
             this.propertyMap = new HashMap<>(properties);
         }
 
+        public <T> boolean hasProp(@Nonnull RecordLayerPropertyKey<T> propKey) {
+            return propertyMap.containsKey(propKey);
+        }
+
+        public <T> void removeProp(@Nonnull RecordLayerPropertyKey<T> propKey) {
+            propertyMap.remove(propKey);
+        }
+
         public <T> Builder addProp(@Nonnull RecordLayerPropertyValue<T> propValue) {
             if (this.propertyMap.putIfAbsent(propValue.getKey(), propValue) != null) {
                 throw new RecordCoreException("Duplicate property name is added")

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. **/

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<Boolean> LUCENE_INDEX_ENCRYPTION_ENABLED = RecordLayerPropertyKey.booleanPropertyKey("com.apple.foundationdb.record.lucene.encryptionEnabled", false);
 
+    public static final RecordLayerPropertyKey<SerializationKeyManager> 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}.
      */

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<ComparablePair<Long, Integer>, CompletableFuture<byte[]>> 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;
@@ -181,6 +178,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;
@@ -229,8 +227,9 @@ private FDBDirectory(@Nonnull Subspace subspace, @Nullable Map<String, String> 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(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;
@@ -403,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(LuceneSerializer.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",
@@ -425,7 +424,8 @@ 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));
+        agilityContext.increment(LuceneEvents.Counts.LUCENE_BLOCK_WRITES);
         //This may not be correct transactionally
         agilityContext.recordSize(LuceneEvents.SizeEvents.LUCENE_WRITE, encodedBytes.length);
         if (LOGGER.isTraceEnabled()) {
@@ -538,7 +538,7 @@ private CompletableFuture<byte[]> readBlock(@Nonnull IndexInput requestingInput,
     private CompletableFuture<byte[]> 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 +624,7 @@ private CompletableFuture<Void> 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))
@@ -912,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 = LuceneSerializer.encode(value.getBytes(), compressionEnabled, encryptionEnabled);
+                byte[] encodedBytes = serializer.encode(value.getBytes());
                 agilityContext.set(metaSubspace.pack(dest), encodedBytes);
                 agilityContext.clear(key);
 
@@ -1037,6 +1037,11 @@ public <T> T asyncToSync(@Nonnull StoreTimer.Wait event, @Nonnull CompletableFut
         return agilityContext.asyncToSync(event, async);
     }
 
+    @Nullable
+    public LuceneSerializer getSerializer() {
+        return serializer;
+    }
+
     public Subspace getSubspace() {
         return subspace;
     }
@@ -1049,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());
     }
 
     /**