Pending write queue infrastructure to support non-trivial indexes

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

@@ -779,7 +779,13 @@ private <M extends Message> void updateSecondaryIndexes(@Nullable final FDBIndex
                 case WRITE_ONLY_WITH_QUEUE:
                     // Push the old/new record to a write pending queue instead of updating the index directly. The
                     // ongoing online indexer will drain the queue and perform the actual index update.
-                    future = maintainer.updateWhileWriteOnlyWithQueue(oldRecord, newRecord);
+                    if (!maintainer.isPendingWriteQueueAllowed()) {
+                        // The indexer should not have allowed the WRITE_ONLY_WITH_QUEUE index state for this maintainer.
+                        throw new RecordCoreException("index does not support the pending write queue")
+                                .addLogInfo(LogMessageKeys.INDEX_NAME, index.getName());
+                    }
+                    future = IndexingPendingWriteQueue.enqueuePendingIndexUpdate(this, index,
+                            maintainer.serializePendingWriteQueue(oldRecord, newRecord));
                     context.addToSessionSet(ContextSessionKey.WRITE_ONLY_WITH_QUEUE_INDEXES_UPDATED, index.getName());
                     break;
 

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

@@ -24,6 +24,7 @@
 import com.apple.foundationdb.annotation.API;
 import com.apple.foundationdb.async.AsyncUtil;
 import com.apple.foundationdb.record.EvaluationContext;
+import com.apple.foundationdb.record.IndexBuildProto;
 import com.apple.foundationdb.record.IndexEntry;
 import com.apple.foundationdb.record.IndexScanType;
 import com.apple.foundationdb.record.IsolationLevel;
@@ -35,7 +36,6 @@
 import com.apple.foundationdb.record.metadata.IndexRecordFunction;
 import com.apple.foundationdb.record.metadata.Key;
 import com.apple.foundationdb.record.provider.foundationdb.indexes.InvalidIndexEntry;
-import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueue;
 import com.apple.foundationdb.record.query.QueryToKeyMatcher;
 import com.apple.foundationdb.subspace.Subspace;
 import com.apple.foundationdb.tuple.Tuple;
@@ -155,22 +155,32 @@ public abstract <M extends Message> CompletableFuture<Void> updateWhileWriteOnly
 
 
     /**
-     * While the index state is in {@link com.apple.foundationdb.record.IndexState#WRITE_ONLY_WITH_QUEUE}, push the information
-     * to a write pending queue. The ongoing online indexer session will later drain the queue and call
-     * {@link #updateWhileWriteOnly(FDBIndexableRecord, FDBIndexableRecord)} with the same parameters.
-     * This index state was designed to prevent repeating conflicts with indexer's transaction when the index
-     * maintainer path includes bottlenecks.
+     * Serialize the old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry} message that can be saved
+     * as an entry in a pending write queue. An ongoing online indexer session will later drain the queue and call
+     * {@link #updateFromQueue(IndexBuildProto.PendingWritesQueueEntry)} with this entry.
+     * <p>
+     * This is only called for maintainers that allow pending write queue (see {@link #isPendingWriteQueueAllowed()}); the
+     * caller is responsible for checking that before invoking this method.
      *
      * @param oldRecord the previous stored record or <code>null</code> if a new record is being created
      * @param newRecord the new record or <code>null</code> if an old record is being deleted
      * @param <M> type of message
-     * @return a future that is complete when the index update is done
-     * @throws PendingWritesQueue.PendingWritesQueueTooLargeException via the returned future if the queue is
-     * oversized. TODO: eliminate this potential exception
+     * @return the queue entry message to enqueue
      */
     @Nonnull
-    public abstract <M extends Message> CompletableFuture<Void> updateWhileWriteOnlyWithQueue(@Nullable FDBIndexableRecord<M> oldRecord,
-                                                                                              @Nullable FDBIndexableRecord<M> newRecord);
+    public abstract <M extends Message> IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable FDBIndexableRecord<M> oldRecord,
+                                                                                                          @Nullable FDBIndexableRecord<M> newRecord);
+
+
+    /**
+     * Apply a queued index update that was previously deferred onto the pending writes queue while the index was in the
+     * {@link com.apple.foundationdb.record.IndexState#WRITE_ONLY_WITH_QUEUE} state.
+     *
+     * @param payload the queued entry to apply
+     * @return a future that is complete when the update has been applied
+     */
+    @Nonnull
+    public abstract CompletableFuture<Void> updateFromQueue(@Nonnull IndexBuildProto.PendingWritesQueueEntry payload);
 
 
     /**
@@ -314,6 +324,16 @@ protected CompletableFuture<Tuple> unsupportedAggregateFunction(@Nonnull IndexAg
      */
     public abstract boolean isIdempotent();
 
+    /**
+     * Whether this index maintainer supports being built with a pending write queue (the
+     * {@link com.apple.foundationdb.record.IndexState#WRITE_ONLY_WITH_QUEUE WRITE_ONLY_WITH_QUEUE} index state). While
+     * an index is in this state, user updates are deferred to a pending write queue instead of being applied to the
+     * index directly, and the online indexer drains that queue as it builds.
+     * Maintainers whose cannot correctly defer and later replay updates should return {@code false} to refuse this state;
+     * @return whether this index may be built with a pending write queue
+     */
+    public abstract boolean isPendingWriteQueueAllowed();
+
     /**
      * Whether this key has been added to some range within the {@link com.apple.foundationdb.async.RangeSet RangeSet}
      * associated with this index. This is used within the context of seeing if one should update a non-idempotent

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

@@ -100,7 +100,7 @@ public abstract class IndexingBase {
     private boolean forceStampOverwrite = false;
     private final long startingTimeMillis;
     private Map<String, IndexingMerger> indexingMergerMap = null;
-    private Map<String, PendingWriteQueueDrainer> indexingDrainerMap = null;
+    private Map<String, IndexingPendingWriteQueue> indexingDrainerMap = null;
     @Nullable
     private IndexingHeartbeat heartbeat = null; // this will stay null for index scrubbing
 
@@ -283,13 +283,12 @@ private CompletableFuture<Boolean> markSingleIndexWriteOnly(final FDBRecordStore
         // drained version-key index would be built with a null version. It is also not allowed during mutual indexing,
         // as two concurrent queue drains may cause data inconsistency.
         final Index index = indexContext.index;
+        final IndexMaintainer maintainer = store.getIndexMaintainer(index);
         if (policy.shouldUsePendingWriteQueue(index) &&
-                !indexContext.isSynthetic &&
                 !policy.isMutual() &&
-                store.getIndexMaintainer(index).isIdempotent() &&
+                maintainer.isPendingWriteQueueAllowed() &&
                 index.getRootExpression().versionColumns() == 0 &&
                 store.getFormatVersionEnum().isAtLeast(FormatVersion.WRITE_ONLY_WITH_QUEUE)) {
-            // TODO: support write-only-with-queue for synthetic records
             // TODO? support versioned index ("?" because these kind of indexes don't tend to have indexing bottlenecks)
             // TODO? support write-only-with-queue for mutual indexing (may require a drain semaphore)
             return store.markIndexWriteOnlyWithQueue(index);
@@ -1079,11 +1078,11 @@ private synchronized IndexingMerger getIndexingMerger(Index index) {
         return indexingMergerMap.computeIfAbsent(index.getName(), k -> new IndexingMerger(index, common, policy.getInitialMergesCountLimit()));
     }
 
-    private synchronized PendingWriteQueueDrainer getIndexingDrainer(Index index) {
+    private synchronized IndexingPendingWriteQueue getIndexingDrainer(Index index) {
         if (indexingDrainerMap == null) {
             indexingDrainerMap = new HashMap<>();
         }
-        return indexingDrainerMap.computeIfAbsent(index.getName(), k -> new PendingWriteQueueDrainer(index, common));
+        return indexingDrainerMap.computeIfAbsent(index.getName(), k -> new IndexingPendingWriteQueue(index, common));
     }
 
     private void deferAutoMergeDuringCommit(FDBRecordStore store) {

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java → fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java

@@ -1,9 +1,9 @@
 /*
- * PendingWriteQueueDrainer.java
+ * IndexingPendingWriteQueue.java
  *
  * This source file is part of the FoundationDB open source project
  *
- * Copyright 2015-2023 Apple Inc. and the FoundationDB project authors
+ * Copyright 2015-2026 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.
@@ -38,28 +38,31 @@
 import java.util.concurrent.CompletableFuture;
 
 /**
- * Drain a pending writes queue. Used by the indexer.
+ * Use {@link PendingWritesQueue} to defer index updates while an index is being built.
+ * Indexing maintainer: will use this module to push items to the queue
+ * Online indexer: will use this module to drain the queue and update the index
  */
 @ParametersAreNonnullByDefault
-public class PendingWriteQueueDrainer {
+public final class IndexingPendingWriteQueue {
+    // TODO: configurable maxQueueSize
+    private static final int MAX_QUEUE_SIZE = 100_000;
+    private static final int MAX_RECORDS_DELETE_PER_SECOND = 10_000;
+
     private final Index index;
     private final IndexingCommon common;
-    private static final int MAX_RECORDS_DELETE_PER_SECOND = 10_000;
 
-    public PendingWriteQueueDrainer(final Index index, IndexingCommon common) {
+    public IndexingPendingWriteQueue(final Index index, final IndexingCommon common) {
         this.index = index;
         this.common = common;
     }
 
     CompletableFuture<Boolean> isQueueEmpty(FDBRecordStore store) {
-        return getQueue(store).isQueueEmpty(store.getContext());
+        return getIndexingQueue(store, index).isQueueEmpty(store.getContext());
     }
 
     @SuppressWarnings("PMD.CloseResource")
     CompletableFuture<Void> drainPendingQueue() {
-        // Propagate the indexer's timer to the drain transactions. Besides preserving metrics, this is required for
-        // index maintainers that assume a non-null timer while applying updates (e.g. the vector/HNSW maintainer),
-        // which would otherwise fail with a NullPointerException as the queued writes are replayed.
+        // Called by the indexer: update the index and then remove every queue item
         final FDBRecordContextConfig.Builder contextConfigBuilder =
                 FDBRecordContextConfig.newBuilder().setTimer(common.getRunner().getTimer());
         final ThrottledRetryingIterator<PendingWritesQueueEntry<IndexBuildProto.PendingWritesQueueEntry>> iterator =
@@ -85,7 +88,7 @@ private CursorFactory<PendingWritesQueueEntry<IndexBuildProto.PendingWritesQueue
         return (store, lastResult, rowLimit) -> {
             final byte[] continuation = lastResult == null ? null : lastResult.getContinuation().toBytes();
             final ScanProperties scanProperties = ScanProperties.FORWARD_SCAN.with(props -> props.setReturnedRowLimit(rowLimit));
-            return getQueue(store).getQueueCursor(store.getContext(), scanProperties, continuation);
+            return getIndexingQueue(store, index).getQueueCursor(store.getContext(), scanProperties, continuation);
         };
     }
 
@@ -99,19 +102,50 @@ private CompletableFuture<Void> handleOneItem(final FDBRecordStore store,
         }
         final IndexBuildProto.PendingWritesQueueEntry payload = entry.getPayload();
         return store.getIndexMaintainer(index)
-                // Calling updateWhileWriteOnly explicitly, lest this update will be re-pushed to the queue
-                .updateWhileWriteOnly(
-                        PendingWriteQueueIndexingFactory.getOldRecord(store, payload),
-                        PendingWriteQueueIndexingFactory.getNewRecord(store, payload))
+                .updateFromQueue(payload)
                 .thenAccept(ignore -> {
                     quotaManager.deleteCountInc();
-                    getQueue(store).clearEntry(store.getContext(), entry);
+                    getIndexingQueue(store, index).clearEntry(store.getContext(), entry);
                 });
     }
 
     @Nonnull
-    private PendingWritesQueue<IndexBuildProto.PendingWritesQueueEntry> getQueue(final FDBRecordStore store) {
-        return PendingWriteQueueIndexingFactory.getIndexingQueue(store, index);
+    public static PendingWritesQueue<IndexBuildProto.PendingWritesQueueEntry> getIndexingQueue(final FDBRecordStore store, final Index index) {
+        return new PendingWritesQueue<>(
+                IndexingSubspaces.indexPendingWriteQueueSubspace(store, index),
+                IndexingSubspaces.indexPendingWriteQueueSizeSubspace(store, index),
+                MAX_QUEUE_SIZE,
+                IndexBuildProto.PendingWritesQueueEntry.class
+        );
+    }
+
+    /**
+     * Return true if the pending writes queue for the given index currently holds at least one entry. The size counter
+     * is read via a snapshot (conflict-free) read.
+     * @param store the record store whose queue is inspected
+     * @param index the index whose queue is inspected
+     * @param context the context used for the conflict-free read
+     * @return a future that completes with true if the queue is non-empty
+     */
+    @Nonnull
+    public static CompletableFuture<Boolean> hasPendingWrites(final FDBRecordStore store, final Index index, final FDBRecordContext context) {
+        return getIndexingQueue(store, index).getQueueSizeNoConflict(context)
+                .thenApply(size -> size != null && size > 0);
+    }
+
+    /**
+     * Called by the index maintainer to enqueue data for a deferred index update.
+     * @param store the record store whose incarnation and context are used
+     * @param index the index whose queue the entry is appended to
+     * @param entry the entry to enqueue
+     * @return a future that completes when the entry has been enqueued
+     */
+    @Nonnull
+    public static CompletableFuture<Void> enqueuePendingIndexUpdate(
+            final FDBRecordStore store,
+            final Index index,
+            final IndexBuildProto.PendingWritesQueueEntry entry) {
+        return getIndexingQueue(store, index).enqueue(store.getContext(), entry, store.getIncarnation());
     }
 
     /**

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

@@ -24,15 +24,13 @@
 import com.apple.foundationdb.FDBException;
 import com.apple.foundationdb.annotation.API;
 import com.apple.foundationdb.async.AsyncUtil;
-import com.apple.foundationdb.record.IndexBuildProto;
 import com.apple.foundationdb.record.IndexState;
 import com.apple.foundationdb.record.RecordCoreStorageException;
 import com.apple.foundationdb.record.logging.KeyValueLogMessage;
 import com.apple.foundationdb.record.logging.LogMessageKeys;
 import com.apple.foundationdb.record.metadata.Index;
 import com.apple.foundationdb.record.provider.common.StoreTimer;
 import com.apple.foundationdb.record.provider.common.StoreTimerSnapshot;
-import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueue;
 import com.apple.foundationdb.record.provider.foundationdb.runners.ExponentialDelay;
 import com.apple.foundationdb.record.util.Result;
 import com.apple.foundationdb.util.LoggableException;
@@ -468,12 +466,10 @@ private CompletableFuture<Void> populateDrainRequiredIndexes(FDBRecordStore stor
      */
     private CompletableFuture<List<Index>> nonEmptyQueueIndexes(@Nonnull FDBRecordStore store, @Nonnull FDBRecordContext context, @Nonnull List<Index> indexes) {
         return AsyncUtil.getAll(indexes.stream()
-                        .map(index -> {
-                            final PendingWritesQueue<IndexBuildProto.PendingWritesQueueEntry> queue =
-                                    PendingWriteQueueIndexingFactory.getIndexingQueue(store, index);
-                            return queue.getQueueSizeNoConflict(context)
-                                    .thenApply(size -> size != null && size > 0 ? index : null);
-                        })
+                        .map(index ->
+                                IndexingPendingWriteQueue.hasPendingWrites(store, index, context)
+                                        .thenApply(hasPending -> hasPending ? index : null)
+                        )
                         .toList()
                 )
                 .thenApply(results -> results.stream().filter(Objects::nonNull).toList());