Lucene: handle requests for deferred merges (#3905)

1. Require deferred merge when queuing a pending write. This is done to
cover cases of a merge indicator without active merge. Requesting a
deferred merge will trigger merge and then drain the queue and clear the
indicator.
2. Do not require a deferred merge in the explicit merge path. This will
prevent an endless loop void merges.
3. As an optimization, use a `NoMergePolicy` in the non-indexing path.

Author: jjezra

fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/LuceneIndexMaintainer.java

@@ -177,6 +177,8 @@ private <M extends Message> void writeDocument(final FDBIndexableRecord<M> newRe
         if (shouldUseQueue(entry.getKey(), partitionId)) {
             PendingWriteQueue queue = directoryManager.getPendingWriteQueue(entry.getKey(), partitionId);
             queue.enqueueInsert(state.context, newRecord.getPrimaryKey(), entry.getValue());
+            // Require deferred merge (+ drain) in case there is a merge indicator without an active merge
+            this.state.store.getIndexDeferredMaintenanceControl().setMergeRequiredIndexes(this.state.index);
         } else {
             writeDocumentBypassQueue(newRecord, entry, partitionId);
         }
@@ -204,6 +206,8 @@ private int deleteDocument(Tuple groupingKey, @Nullable Integer partitionId, Tup
         if (shouldUseQueue(groupingKey, partitionId)) {
             PendingWriteQueue queue = directoryManager.getPendingWriteQueue(groupingKey, partitionId);
             queue.enqueueDelete(state.context, primaryKey);
+            // Require deferred merge (+ drain) in case there is a merge indicator without an active merge
+            this.state.store.getIndexDeferredMaintenanceControl().setMergeRequiredIndexes(this.state.index);
             return 0; // partition count will be adjusted during drain
         } else {
             return deleteDocumentBypassQueue(groupingKey, partitionId, primaryKey);

fdb-record-layer-lucene/src/main/java/com/apple/foundationdb/record/lucene/directory/FDBDirectoryWrapper.java

@@ -56,8 +56,8 @@
 import org.apache.lucene.index.MergePolicy;
 import org.apache.lucene.index.MergeScheduler;
 import org.apache.lucene.index.MergeTrigger;
+import org.apache.lucene.index.NoMergePolicy;
 import org.apache.lucene.index.StandardDirectoryReaderOptimization;
-import org.apache.lucene.index.TieredMergePolicy;
 import org.apache.lucene.store.LockFactory;
 import org.apache.lucene.store.NoLockFactory;
 import org.apache.lucene.util.IOUtils;
@@ -198,21 +198,27 @@ private IndexWriter createIndexWriter(final Exception exceptionAtCreation) throw
     @Nonnull
     private IndexWriterConfig createIndexWriterConfig(final Exception exceptionAtCreation) {
         final IndexDeferredMaintenanceControl mergeControl = this.state.store.getIndexDeferredMaintenanceControl();
-        TieredMergePolicy tieredMergePolicy = new FDBTieredMergePolicy(mergeControl, this.agilityContext,
-                this.state.indexSubspace, this.key, exceptionAtCreation)
-                .setMaxMergedSegmentMB(this.state.context.getPropertyStorage().getPropertyValue(LuceneRecordContextProperties.LUCENE_MERGE_MAX_SIZE))
-                .setSegmentsPerTier(this.state.context.getPropertyStorage().getPropertyValue(LuceneRecordContextProperties.LUCENE_MERGE_SEGMENTS_PER_TIER));
-        tieredMergePolicy.setNoCFSRatio(1.00);
-        IndexWriterConfig indexWriterConfig = new IndexWriterConfig(this.analyzerWrapper.getAnalyzer())
+        final MergePolicy mergePolicy;
+        if (mergeControl.shouldAutoMergeDuringCommit() || mergeControl.isExplicitMergePath()) {
+            // Here: this function was called in an explicit merge path. Prepare an appropriate
+            // merge policy, and avoid requesting a deferred merge
+            mergePolicy = new FDBTieredMergePolicy(mergeControl, this.agilityContext,
+                    this.state.indexSubspace, this.key, exceptionAtCreation)
+                    .setMaxMergedSegmentMB(this.state.context.getPropertyStorage().getPropertyValue(LuceneRecordContextProperties.LUCENE_MERGE_MAX_SIZE))
+                    .setSegmentsPerTier(this.state.context.getPropertyStorage().getPropertyValue(LuceneRecordContextProperties.LUCENE_MERGE_SEGMENTS_PER_TIER));
+        } else {
+            // Here: Not a merge path, optimize by using a "no merge" policy and request a deferred merge
+            mergePolicy = NoMergePolicy.INSTANCE;
+            mergeControl.setMergeRequiredIndexes(this.state.index);
+        }
+
+        mergePolicy.setNoCFSRatio(1.00);
+        return new IndexWriterConfig(this.analyzerWrapper.getAnalyzer())
                 .setUseCompoundFile(USE_COMPOUND_FILE)
-                .setMergePolicy(tieredMergePolicy)
+                .setMergePolicy(mergePolicy)
                 .setMergeScheduler(getMergeScheduler(this.state, this.mergeDirectoryCount, this.agilityContext, this.key))
                 .setCodec(CODEC)
                 .setInfoStream(new LuceneLoggerInfoStream(LOGGER));
-
-        // Merge is required when creating an index writer (do we have a better indicator for a required merge?)
-        mergeControl.setMergeRequiredIndexes(this.state.index);
-        return indexWriterConfig;
     }
 
     @Nonnull

fdb-record-layer-lucene/src/test/java/com/apple/foundationdb/record/lucene/directory/PendingWriteQueueTest.java

@@ -942,6 +942,56 @@ void pendingQueueTestDrainException2() {
         });
     }
 
+    @Test
+    void testMergeRequiredIndexesWithPendingQueue() {
+        // Test that store's getMergeRequiredIndexes does the right thing
+        final Index index = SIMPLE_TEXT_SUFFIXES;
+        final KeySpacePath path = pathManager.createPath(TestKeySpace.RECORD_STORE);
+        final Function<FDBRecordContext, FDBRecordStore> schemaSetup = context ->
+                LuceneIndexTestUtils.rebuildIndexMetaData(context, path,
+                        TestRecordsTextProto.SimpleDocument.getDescriptor().getName(),
+                        index, useCascadesPlanner).getLeft();
+
+        // Insert without queue indicator - should request deferred merge
+        try (FDBRecordContext context = openContext()) {
+            FDBRecordStore recordStore = Objects.requireNonNull(schemaSetup.apply(context));
+            recordStore.saveRecord(LuceneIndexTestUtils.createSimpleDocument(1001L, "first document", 1));
+
+            Set<Index> mergeRequired = recordStore.getIndexDeferredMaintenanceControl().getMergeRequiredIndexes();
+            assertNotNull(mergeRequired);
+            assertTrue(mergeRequired.contains(index));
+
+            commit(context);
+        }
+
+        // Enable queue mode
+        setOngoingMergeIndicator(schemaSetup, index, null, null);
+
+        // Insert with queue indicator - should request deferred merge
+        try (FDBRecordContext context = openContext()) {
+            FDBRecordStore recordStore = Objects.requireNonNull(schemaSetup.apply(context));
+            recordStore.saveRecord(LuceneIndexTestUtils.createSimpleDocument(1002L, "second document", 1));
+
+            Set<Index> mergeRequired = recordStore.getIndexDeferredMaintenanceControl().getMergeRequiredIndexes();
+            assertNotNull(mergeRequired);
+            assertTrue(mergeRequired.contains(index));
+
+            commit(context);
+        }
+
+        // Call explicit merge - should not request deferred merge
+        try (FDBRecordContext context = openContext()) {
+            FDBRecordStore recordStore = Objects.requireNonNull(schemaSetup.apply(context));
+            final LuceneIndexMaintainer indexMaintainer = getIndexMaintainer(recordStore, index);
+            indexMaintainer.mergeIndex().join();
+
+            Set<Index> mergeRequired = recordStore.getIndexDeferredMaintenanceControl().getMergeRequiredIndexes();
+            assertTrue(mergeRequired == null || mergeRequired.isEmpty());
+
+            commit(context);
+        }
+    }
+
     private void verifyClearedQueueAndIndicator(Function<FDBRecordContext, FDBRecordStore> schemaSetup, Index index, @Nullable Tuple groupingKey, @Nullable Integer partitionId) {
         try (FDBRecordContext context = openContext()) {
             FDBRecordStore recordStore = Objects.requireNonNull(schemaSetup.apply(context));