From 56d4fe48866688c33649851806c500ce3e064201 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Mon, 13 Jul 2026 18:03:31 -0400 Subject: [PATCH 01/15] wip: synethtic in the standard index maintainer --- .../foundationdb/indexes/StandardIndexMaintainer.java | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index 545d780e47..76cdec3990 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -339,9 +339,18 @@ public CompletableFuture updateWhileWriteOnlyWithQueue // had generated an "already indexed" source index key, the other one should be converted to a null in the queue). // This, however can be done at a later step. } + if (!isSynthtic()) { + throw new UnsupportedOperationException("synthetic records indexes cannot support this operation through the standard maintainer"); + // To support synthetic records, the specific index maintainer should override this function and save + // oldRecord, newRecords with its own serializer + } return PendingWriteQueueIndexingFactory.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); } + private boolean isSynthtic() { + return state.store.getRecordMetaData().recordTypesForIndex(state.index).stream().anyMatch(RecordType::isSynthetic); + } + @Nullable private static Tuple evaluateSingletonIndexKey(Index index, IndexMaintainer maintainer, @Nullable FDBIndexableRecord record) { if (record == null) { From ea12c47fc1c2d30e7631f7f344f88a3caf5d2a67 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Mon, 13 Jul 2026 19:14:08 -0400 Subject: [PATCH 02/15] Maintainers infra isPendingWriteQueueAllowed --- .../provider/foundationdb/IndexMaintainer.java | 11 +++++++++++ .../provider/foundationdb/IndexingBase.java | 6 +++--- .../indexes/NoOpIndexMaintainer.java | 5 +++++ .../indexes/SlidingWindowIndexMaintainer.java | 5 +++++ .../indexes/StandardIndexMaintainer.java | 16 +++++++++------- .../FDBRecordStoreUniqueIndexTest.java | 5 +++++ .../foundationdb/OnlineIndexerMergeTest.java | 5 +++++ .../foundationdb/TerribleIndexMaintainer.java | 5 +++++ .../indexes/SlidingWindowIndexTest.java | 5 +++++ 9 files changed, 53 insertions(+), 10 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java index 5e8006a24e..ac87b755b6 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java @@ -314,6 +314,17 @@ protected CompletableFuture 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 {@link #updateWhileWriteOnlyWithQueue} 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 diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java index 9157edc384..c2253ed3c4 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java @@ -283,13 +283,13 @@ private CompletableFuture 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.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); diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java index 4f747d9105..8d1c6a49ae 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java @@ -137,6 +137,11 @@ public boolean isIdempotent() { return false; } + @Override + public boolean isPendingWriteQueueAllowed() { + return false; + } + @Nonnull @Override public CompletableFuture addedRangeWithKey(@Nonnull Tuple primaryKey) { diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java index 88f28a2d8d..8ebdfe829b 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java @@ -361,6 +361,11 @@ public boolean isIdempotent() { return delegate.isIdempotent(); } + @Override + public boolean isPendingWriteQueueAllowed() { + return delegate.isPendingWriteQueueAllowed(); + } + @Nonnull @Override public CompletableFuture update(@Nullable FDBIndexableRecord oldRecord, diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index 76cdec3990..ac528828e5 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -339,16 +339,12 @@ public CompletableFuture updateWhileWriteOnlyWithQueue // had generated an "already indexed" source index key, the other one should be converted to a null in the queue). // This, however can be done at a later step. } - if (!isSynthtic()) { - throw new UnsupportedOperationException("synthetic records indexes cannot support this operation through the standard maintainer"); - // To support synthetic records, the specific index maintainer should override this function and save - // oldRecord, newRecords with its own serializer - } return PendingWriteQueueIndexingFactory.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); } - private boolean isSynthtic() { - return state.store.getRecordMetaData().recordTypesForIndex(state.index).stream().anyMatch(RecordType::isSynthetic); + private boolean isSynthetic() { + return !state.store.getRecordMetaData().getSyntheticRecordTypes().isEmpty() && + state.store.getRecordMetaData().recordTypesForIndex(state.index).stream().anyMatch(RecordType::isSynthetic); } @Nullable @@ -759,6 +755,12 @@ public boolean isIdempotent() { return true; } + @Override + public boolean isPendingWriteQueueAllowed() { + // Synthetic records index maintainers that support pending write queues should override this function + return !isSynthetic(); + } + @Override @Nonnull public CompletableFuture addedRangeWithKey(@Nonnull Tuple primaryKey) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java index 320d3e9b73..4a6f2a5476 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java @@ -912,6 +912,11 @@ public boolean isIdempotent() { return underlying.isIdempotent(); } + @Override + public boolean isPendingWriteQueueAllowed() { + return underlying.isPendingWriteQueueAllowed(); + } + @Nonnull @Override public CompletableFuture addedRangeWithKey(@Nonnull final Tuple primaryKey) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java index 45e94bc4b1..3d35fb6d6e 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java @@ -489,6 +489,11 @@ public boolean isIdempotent() { throw new UnsupportedOperationException(); } + @Override + public boolean isPendingWriteQueueAllowed() { + throw new UnsupportedOperationException(); + } + @Nonnull @Override public CompletableFuture addedRangeWithKey(@Nonnull final Tuple primaryKey) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java index de736d8047..8dfe6b01b3 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java @@ -149,6 +149,11 @@ public boolean isIdempotent() { return false; } + @Override + public boolean isPendingWriteQueueAllowed() { + return false; + } + @Nonnull @Override public CompletableFuture addedRangeWithKey(@Nonnull Tuple primaryKey) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java index 94a5908ebc..5ba3053fda 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java @@ -1591,6 +1591,11 @@ public boolean isIdempotent() { return true; } + @Override + public boolean isPendingWriteQueueAllowed() { + return true; + } + @Nonnull @Override public CompletableFuture addedRangeWithKey(@Nonnull Tuple primaryKey) { From cd35415da83f824ad8c2c486930b5b701550a35b Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 11:45:54 -0400 Subject: [PATCH 03/15] Rename PendingWriteQueueIndexingFactory.java -> IndexingPendingWriteQueue.java --- .../record/provider/foundationdb/IndexingBase.java | 1 - ...eIndexingFactory.java => IndexingPendingWriteQueue.java} | 6 +++--- .../record/provider/foundationdb/IndexingThrottle.java | 2 +- .../provider/foundationdb/PendingWriteQueueDrainer.java | 6 +++--- .../foundationdb/indexes/SlidingWindowIndexMaintainer.java | 4 ++-- .../foundationdb/indexes/StandardIndexMaintainer.java | 6 +++--- .../provider/foundationdb/FDBRecordStoreIndexTest.java | 4 ++-- .../foundationdb/OnlineIndexerPendingWriteQueueTest.java | 6 +++--- .../foundationdb/indexes/SlidingWindowIndexTest.java | 4 ++-- 9 files changed, 19 insertions(+), 20 deletions(-) rename fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/{PendingWriteQueueIndexingFactory.java => IndexingPendingWriteQueue.java} (97%) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java index c2253ed3c4..7fd23e0ae0 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java @@ -286,7 +286,6 @@ private CompletableFuture markSingleIndexWriteOnly(final FDBRecordStore final IndexMaintainer maintainer = store.getIndexMaintainer(index); if (policy.shouldUsePendingWriteQueue(index) && !policy.isMutual() && - maintainer.isIdempotent() && maintainer.isPendingWriteQueueAllowed() && index.getRootExpression().versionColumns() == 0 && store.getFormatVersionEnum().isAtLeast(FormatVersion.WRITE_ONLY_WITH_QUEUE)) { diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueIndexingFactory.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java similarity index 97% rename from fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueIndexingFactory.java rename to fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index 2d0e14efb9..c550315106 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueIndexingFactory.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -1,5 +1,5 @@ /* - * PendingWriteQueueIndexingFactory.java + * IndexingPendingWriteQueue.java * * This source file is part of the FoundationDB open source project * @@ -43,11 +43,11 @@ * and consumers (the drainer) always agree on it. */ @ParametersAreNonnullByDefault -public final class PendingWriteQueueIndexingFactory { +public final class IndexingPendingWriteQueue { // TODO: configurable maxQueueSize private static final int MAX_QUEUE_SIZE = 100_000; - private PendingWriteQueueIndexingFactory() { + private IndexingPendingWriteQueue() { } @Nonnull diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java index e654ab8468..1fa5aa2391 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java @@ -470,7 +470,7 @@ private CompletableFuture> nonEmptyQueueIndexes(@Nonnull FDBRecordSt return AsyncUtil.getAll(indexes.stream() .map(index -> { final PendingWritesQueue queue = - PendingWriteQueueIndexingFactory.getIndexingQueue(store, index); + IndexingPendingWriteQueue.getIndexingQueue(store, index); return queue.getQueueSizeNoConflict(context) .thenApply(size -> size != null && size > 0 ? index : null); }) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java index a92b419029..4f3b019fc7 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java @@ -101,8 +101,8 @@ private CompletableFuture handleOneItem(final FDBRecordStore store, 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)) + IndexingPendingWriteQueue.getOldRecord(store, payload), + IndexingPendingWriteQueue.getNewRecord(store, payload)) .thenAccept(ignore -> { quotaManager.deleteCountInc(); getQueue(store).clearEntry(store.getContext(), entry); @@ -111,7 +111,7 @@ private CompletableFuture handleOneItem(final FDBRecordStore store, @Nonnull private PendingWritesQueue getQueue(final FDBRecordStore store) { - return PendingWriteQueueIndexingFactory.getIndexingQueue(store, index); + return IndexingPendingWriteQueue.getIndexingQueue(store, index); } /** diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java index 8ebdfe829b..b640f8df3b 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java @@ -53,7 +53,7 @@ import com.apple.foundationdb.record.provider.foundationdb.IndexOperation; import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; import com.apple.foundationdb.record.logging.LogMessageKeys; -import com.apple.foundationdb.record.provider.foundationdb.PendingWriteQueueIndexingFactory; +import com.apple.foundationdb.record.provider.foundationdb.IndexingPendingWriteQueue; import com.apple.foundationdb.record.query.QueryToKeyMatcher; import com.apple.foundationdb.subspace.Subspace; import com.apple.foundationdb.tuple.ByteArrayUtil; @@ -431,7 +431,7 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - return PendingWriteQueueIndexingFactory.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); + return IndexingPendingWriteQueue.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); } /** diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index ac528828e5..482fed562e 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -65,8 +65,8 @@ import com.apple.foundationdb.record.provider.foundationdb.IndexPrefetchRangeKeyValueCursor; import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; import com.apple.foundationdb.record.provider.foundationdb.IndexScanRange; +import com.apple.foundationdb.record.provider.foundationdb.IndexingPendingWriteQueue; import com.apple.foundationdb.record.provider.foundationdb.KeyValueCursor; -import com.apple.foundationdb.record.provider.foundationdb.PendingWriteQueueIndexingFactory; import com.apple.foundationdb.record.provider.foundationdb.indexing.IndexingRangeSet; import com.apple.foundationdb.record.query.QueryToKeyMatcher; import com.apple.foundationdb.subspace.Subspace; @@ -339,7 +339,7 @@ public CompletableFuture updateWhileWriteOnlyWithQueue // had generated an "already indexed" source index key, the other one should be converted to a null in the queue). // This, however can be done at a later step. } - return PendingWriteQueueIndexingFactory.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); + return IndexingPendingWriteQueue.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); } private boolean isSynthetic() { @@ -758,7 +758,7 @@ public boolean isIdempotent() { @Override public boolean isPendingWriteQueueAllowed() { // Synthetic records index maintainers that support pending write queues should override this function - return !isSynthetic(); + return isIdempotent() && !isSynthetic(); } @Override diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java index f13d19573f..b0eee56b58 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java @@ -1000,7 +1000,7 @@ void writeOnlyWithQueueIndex() throws Exception { // The StandardIndexMaintainer routed the update to its pending queue: exactly one deferred write. final PendingWritesQueue standardQueue = - PendingWriteQueueIndexingFactory.getIndexingQueue(recordStore, standardIndex); + IndexingPendingWriteQueue.getIndexingQueue(recordStore, standardIndex); assertThat(standardQueue.getQueueSizeNoConflict(context).join(), is(1L)); // That deferred entry carries the saved record as an insert (a new record, no old record) - i.e. the @@ -1023,7 +1023,7 @@ void writeOnlyWithQueueIndex() throws Exception { // The NoOp-backed permissive index's updateWhileWriteOnlyWithQueue is a no-op, so nothing is enqueued // for it (its queue size counter was never written). - assertThat(PendingWriteQueueIndexingFactory.getIndexingQueue(recordStore, permissiveIndex) + assertThat(IndexingPendingWriteQueue.getIndexingQueue(recordStore, permissiveIndex) .getQueueSizeNoConflict(context).join(), is(nullValue())); commit(context); diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java index 966edbed21..957476b8b4 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java @@ -115,7 +115,7 @@ void testDrainPendingQueueWhileBuilding(int limit) throws Exception { // The pending writes queue should now hold exactly the deferred records (and nothing else), before the drain. try (FDBRecordContext context = openContext()) { final PendingWritesQueue queue = - PendingWriteQueueIndexingFactory.getIndexingQueue(recordStore, index); + IndexingPendingWriteQueue.getIndexingQueue(recordStore, index); final Long queueSize = queue.getQueueSizeNoConflict(context).join(); assertEquals(queuedRecNos.size(), queueSize == null ? 0L : queueSize); @@ -180,7 +180,7 @@ void testDrainPendingQueueWhileBuildingAfterNPasses(final int passesCount) throw // The pending writes queue should now hold exactly the deferred records (and nothing else), before the drain. try (FDBRecordContext context = openContext()) { final PendingWritesQueue queue = - PendingWriteQueueIndexingFactory.getIndexingQueue(recordStore, index); + IndexingPendingWriteQueue.getIndexingQueue(recordStore, index); final Long queueSize = queue.getQueueSizeNoConflict(context).join(); assertEquals(queuedRecNos.size(), queueSize == null ? 0L : queueSize); @@ -1198,7 +1198,7 @@ private List indexPrimaryKeysForValue(@Nonnull final Index index, final in */ private Long queueSizeCounter(@Nonnull final Index index) { try (FDBRecordContext context = openContext()) { - final Long size = PendingWriteQueueIndexingFactory.getIndexingQueue(recordStore, index) + final Long size = IndexingPendingWriteQueue.getIndexingQueue(recordStore, index) .getQueueSizeNoConflict(context).join(); context.commit(); return size; diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java index 5ba3053fda..ee11318507 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java @@ -49,8 +49,8 @@ import com.apple.foundationdb.record.provider.foundationdb.IndexOperation; import com.apple.foundationdb.record.provider.foundationdb.IndexOperationResult; import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; +import com.apple.foundationdb.record.provider.foundationdb.IndexingPendingWriteQueue; import com.apple.foundationdb.record.provider.foundationdb.OnlineIndexer; -import com.apple.foundationdb.record.provider.foundationdb.PendingWriteQueueIndexingFactory; import com.apple.foundationdb.record.provider.foundationdb.VectorIndexScanBounds; import com.apple.foundationdb.record.provider.foundationdb.VectorIndexScanOptions; import com.apple.foundationdb.record.provider.foundationdb.indexes.SlidingWindowTestHelpers.SlidingWindow; @@ -1830,7 +1830,7 @@ void writeOnlyWithQueueRoutesOutOfWindowUpdatesToQueue() throws Exception { @Nullable private Long queueSize(@Nonnull FDBRecordContext context, @Nonnull Index index) { final PendingWritesQueue queue = - PendingWriteQueueIndexingFactory.getIndexingQueue(recordStore, index); + IndexingPendingWriteQueue.getIndexingQueue(recordStore, index); return queue.getQueueSizeNoConflict(context).join(); } } From d3d12e610a9b80c8f6f7fdbaa0a0e0ff16644700 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 12:07:24 -0400 Subject: [PATCH 04/15] Move PendingWriteQueueDrainer functionality to IndexingPendingWriteQueue --- .../provider/foundationdb/IndexingBase.java | 6 +- .../IndexingPendingWriteQueue.java | 96 +++++++++++-- .../PendingWriteQueueDrainer.java | 129 ------------------ .../OnlineIndexerPendingWriteQueueTest.java | 4 +- 4 files changed, 92 insertions(+), 143 deletions(-) delete mode 100644 fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java index 7fd23e0ae0..11a5d6fafc 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java +++ b/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 indexingMergerMap = null; - private Map indexingDrainerMap = null; + private Map indexingDrainerMap = null; @Nullable private IndexingHeartbeat heartbeat = null; // this will stay null for index scrubbing @@ -1078,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) { diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index c550315106..fd3405f761 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -22,32 +22,101 @@ import com.apple.foundationdb.async.AsyncUtil; import com.apple.foundationdb.record.IndexBuildProto; +import com.apple.foundationdb.record.RecordCoreException; +import com.apple.foundationdb.record.RecordCursorResult; import com.apple.foundationdb.record.RecordMetaData; +import com.apple.foundationdb.record.ScanProperties; import com.apple.foundationdb.record.metadata.Index; import com.apple.foundationdb.record.metadata.RecordType; import com.apple.foundationdb.record.provider.common.RecordSerializer; import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueue; +import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueueEntry; +import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.CursorFactory; +import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.ThrottledRetryingIterator; import com.apple.foundationdb.tuple.TupleHelpers; +import com.apple.foundationdb.util.CloseException; import com.google.protobuf.ByteString; import com.google.protobuf.Message; import javax.annotation.Nonnull; import javax.annotation.Nullable; import javax.annotation.ParametersAreNonnullByDefault; +import java.io.Serial; import java.util.concurrent.CompletableFuture; /** - * Factory for the {@link PendingWritesQueue} used by the online indexer to defer index updates while an index is - * being built in the {@link com.apple.foundationdb.record.IndexState#WRITE_ONLY_WITH_QUEUE} state. This keeps the - * indexing-specific queue wiring (subspaces, payload type, capacity) in one place so producers (the index maintainer) - * and consumers (the drainer) always agree on it. + * Use {@link PendingWritesQueue} to defer index updates while an index is being built. + * Indexin 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 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 IndexingPendingWriteQueue() { + private final Index index; + private final IndexingCommon common; + + public IndexingPendingWriteQueue(final Index index, final IndexingCommon common) { + this.index = index; + this.common = common; + } + + CompletableFuture isQueueEmpty(FDBRecordStore store) { + return getIndexingQueue(store, index).isQueueEmpty(store.getContext()); + } + + @SuppressWarnings("PMD.CloseResource") + CompletableFuture drainPendingQueue() { + // 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> iterator = + ThrottledRetryingIterator.builder( + common.getRunner().getDatabase(), + contextConfigBuilder, + cursorFactory(), + this::handleOneItem) + .withMaxRecordsDeletesPerSec(MAX_RECORDS_DELETE_PER_SECOND) + .build(); + return iterator.iterateAll(common.getRecordStoreBuilder().copyBuilder()) + .whenComplete((v, e) -> { + try { + iterator.close(); + } catch (CloseException closeEx) { + throw new PendingWriteQueueDrainException(closeEx); + } + }); + } + + @Nonnull + private CursorFactory> cursorFactory() { + 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 getIndexingQueue(store, index).getQueueCursor(store.getContext(), scanProperties, continuation); + }; + } + + @Nonnull + private CompletableFuture handleOneItem(final FDBRecordStore store, + final RecordCursorResult> lastResult, + final ThrottledRetryingIterator.QuotaManager quotaManager) { + final PendingWritesQueueEntry entry = lastResult.get(); + if (entry == null) { + return AsyncUtil.DONE; + } + final IndexBuildProto.PendingWritesQueueEntry payload = entry.getPayload(); + return store.getIndexMaintainer(index) + // Calling updateWhileWriteOnly explicitly, lest this update will be re-pushed to the queue + .updateWhileWriteOnly( + getOldRecord(store, payload), + getNewRecord(store, payload)) + .thenAccept(ignore -> { + quotaManager.deleteCountInc(); + getIndexingQueue(store, index).clearEntry(store.getContext(), entry); + }); } @Nonnull @@ -61,10 +130,7 @@ public static PendingWritesQueue getInd } /** - * Serialize an old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry} and enqueue it for later - * draining. Either record may be null: a null {@code oldRecord} represents an insert, and a null {@code newRecord} - * represents a delete. This keeps the queue's payload construction next to its other wiring, so index maintainers - * only decide whether to defer an update, not how it is encoded. + * Called by the index maintainer to serialize an old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry}. * @param store the record store whose serializer, incarnation, and context are used * @param index the index whose queue the entry is appended to * @param oldRecord the record prior to the update, or null for an insert @@ -126,4 +192,16 @@ private static FDBStoredRecord deserializeRecord(final FDBRecordStore s return builder.build(); } + /** + * thrown if pending queue drain had failed. + */ + @SuppressWarnings("java:S110") + public static class PendingWriteQueueDrainException extends RecordCoreException { + @Serial + private static final long serialVersionUID = 7; + + public PendingWriteQueueDrainException(final Throwable cause) { + super("Pending write queue drain had failed", cause); + } + } } diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java deleted file mode 100644 index 4f3b019fc7..0000000000 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * PendingWriteQueueDrainer.java - * - * This source file is part of the FoundationDB open source project - * - * Copyright 2015-2023 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.foundationdb; - -import com.apple.foundationdb.async.AsyncUtil; -import com.apple.foundationdb.record.IndexBuildProto; -import com.apple.foundationdb.record.RecordCoreException; -import com.apple.foundationdb.record.RecordCursorResult; -import com.apple.foundationdb.record.ScanProperties; -import com.apple.foundationdb.record.metadata.Index; -import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueue; -import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueueEntry; -import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.CursorFactory; -import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.ThrottledRetryingIterator; -import com.apple.foundationdb.util.CloseException; - -import javax.annotation.Nonnull; -import javax.annotation.ParametersAreNonnullByDefault; -import java.io.Serial; -import java.util.concurrent.CompletableFuture; - -/** - * Drain a pending writes queue. Used by the indexer. - */ -@ParametersAreNonnullByDefault -public class PendingWriteQueueDrainer { - 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) { - this.index = index; - this.common = common; - } - - CompletableFuture isQueueEmpty(FDBRecordStore store) { - return getQueue(store).isQueueEmpty(store.getContext()); - } - - @SuppressWarnings("PMD.CloseResource") - CompletableFuture 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. - final FDBRecordContextConfig.Builder contextConfigBuilder = - FDBRecordContextConfig.newBuilder().setTimer(common.getRunner().getTimer()); - final ThrottledRetryingIterator> iterator = - ThrottledRetryingIterator.builder( - common.getRunner().getDatabase(), - contextConfigBuilder, - cursorFactory(), - this::handleOneItem) - .withMaxRecordsDeletesPerSec(MAX_RECORDS_DELETE_PER_SECOND) - .build(); - return iterator.iterateAll(common.getRecordStoreBuilder().copyBuilder()) - .whenComplete((v, e) -> { - try { - iterator.close(); - } catch (CloseException closeEx) { - throw new PendingWriteQueueDrainException(closeEx); - } - }); - } - - @Nonnull - private CursorFactory> cursorFactory() { - 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); - }; - } - - @Nonnull - private CompletableFuture handleOneItem(final FDBRecordStore store, - final RecordCursorResult> lastResult, - final ThrottledRetryingIterator.QuotaManager quotaManager) { - final PendingWritesQueueEntry entry = lastResult.get(); - if (entry == null) { - return AsyncUtil.DONE; - } - final IndexBuildProto.PendingWritesQueueEntry payload = entry.getPayload(); - return store.getIndexMaintainer(index) - // Calling updateWhileWriteOnly explicitly, lest this update will be re-pushed to the queue - .updateWhileWriteOnly( - IndexingPendingWriteQueue.getOldRecord(store, payload), - IndexingPendingWriteQueue.getNewRecord(store, payload)) - .thenAccept(ignore -> { - quotaManager.deleteCountInc(); - getQueue(store).clearEntry(store.getContext(), entry); - }); - } - - @Nonnull - private PendingWritesQueue getQueue(final FDBRecordStore store) { - return IndexingPendingWriteQueue.getIndexingQueue(store, index); - } - - /** - * thrown if pending queue drain had failed. - */ - @SuppressWarnings("java:S110") - public static class PendingWriteQueueDrainException extends RecordCoreException { - @Serial - private static final long serialVersionUID = 7; - - public PendingWriteQueueDrainException(final Throwable cause) { - super("Pending write queue drain had failed", cause); - } - } -} diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java index 957476b8b4..3ec09ce7d3 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java @@ -890,8 +890,8 @@ void testPendingWriteQueueNotEmptyWhileMarkingReadableExceptionCarriesIndexName( @Test void testPendingWriteQueueDrainExceptionWrapsCause() { final CloseException cause = new CloseException(new RuntimeException("boom")); - final PendingWriteQueueDrainer.PendingWriteQueueDrainException ex = - new PendingWriteQueueDrainer.PendingWriteQueueDrainException(cause); + final IndexingPendingWriteQueue.PendingWriteQueueDrainException ex = + new IndexingPendingWriteQueue.PendingWriteQueueDrainException(cause); assertEquals("Pending write queue drain had failed", ex.getMessage()); assertSame(cause, ex.getCause()); } From 62da19ab9ad0b5fa63a0f4a3920dfd6025954556 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 12:46:51 -0400 Subject: [PATCH 05/15] Convert PendingWritesQueueEntry to a union message --- .../IndexingPendingWriteQueue.java | 19 +++++++++++++------ .../src/main/proto/index_build.proto | 18 +++++++++++------- .../foundationdb/FDBRecordStoreIndexTest.java | 7 ++++--- .../OnlineIndexerPendingWriteQueueTest.java | 4 ++-- 4 files changed, 30 insertions(+), 18 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index fd3405f761..8db464fbde 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -149,14 +149,19 @@ public static CompletableFuture enqueueOldAndNewRecord return AsyncUtil.DONE; } final RecordSerializer serializer = store.getSerializer(); - final IndexBuildProto.PendingWritesQueueEntry.Builder builder = IndexBuildProto.PendingWritesQueueEntry.newBuilder(); + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords.Builder recordsBuilder = + IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords.newBuilder(); if (oldRecord != null) { - builder.setOldRecords(serializeRecord(store, oldRecord, serializer)); + recordsBuilder.setOldRecords(serializeRecord(store, oldRecord, serializer)); } if (newRecord != null) { - builder.setNewRecord(serializeRecord(store, newRecord, serializer)); + recordsBuilder.setNewRecord(serializeRecord(store, newRecord, serializer)); } - return getIndexingQueue(store, index).enqueue(store.getContext(), builder.build(), store.getIncarnation()); + final IndexBuildProto.PendingWritesQueueEntry entry = + IndexBuildProto.PendingWritesQueueEntry.newBuilder() + .setOldAndNewRecords(recordsBuilder) + .build(); + return getIndexingQueue(store, index).enqueue(store.getContext(), entry, store.getIncarnation()); } @Nonnull @@ -169,12 +174,14 @@ private static ByteString serializeRecord(final FDBRecordSto @Nullable public static FDBStoredRecord getOldRecord(final FDBRecordStore store, IndexBuildProto.PendingWritesQueueEntry payload) { - return payload.hasOldRecords() ? deserializeRecord(store, payload.getOldRecords()) : null; + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); + return records.hasOldRecords() ? deserializeRecord(store, records.getOldRecords()) : null; } @Nullable public static FDBStoredRecord getNewRecord(final FDBRecordStore store, IndexBuildProto.PendingWritesQueueEntry payload) { - return payload.hasNewRecord() ? deserializeRecord(store, payload.getNewRecord()) : null; + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); + return records.hasNewRecord() ? deserializeRecord(store, records.getNewRecord()) : null; } /** diff --git a/fdb-record-layer-core/src/main/proto/index_build.proto b/fdb-record-layer-core/src/main/proto/index_build.proto index 1955faab6f..da09e01ece 100644 --- a/fdb-record-layer-core/src/main/proto/index_build.proto +++ b/fdb-record-layer-core/src/main/proto/index_build.proto @@ -23,6 +23,8 @@ syntax = "proto2"; package com.apple.foundationdb.record; option java_outer_classname = "IndexBuildProto"; +import "record_metadata_options.proto"; + // This stamp indicates an ongoing indexing process. It will be used to assure a coherent indexing // continuation - which continued by another process. @@ -55,12 +57,14 @@ message IndexBuildHeartbeat { } // The message that gets saved into the pending writes queue. -// This message gets stored in an google.protobuf.Any field in the queue. This means that this message type SHOULD NOT be moved -// or renamed to preserve wire-format compatibility with existing data -// This represents a record change that needs to be indexed: -// Either of the records (but not both) can be null, where null "old_record" means we are inserting a new -// record and null "new_record" means we are deleting an existing one. message PendingWritesQueueEntry { - optional bytes old_records = 1; - optional bytes new_record = 2; + option (com.apple.foundationdb.record.record).usage = UNION; + + message OldAndNewRecords { + // Simple cases where store serializer can handle the records (typically this is not the case for synthetic records) + optional bytes old_records = 1; + optional bytes new_record = 2; + } + + optional OldAndNewRecords oldAndNewRecords = 1; } diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java index b0eee56b58..2e068a5968 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java @@ -1008,11 +1008,12 @@ void writeOnlyWithQueueIndex() throws Exception { final IndexBuildProto.PendingWritesQueueEntry payload = standardQueue.getQueueCursor(context, ScanProperties.FORWARD_SCAN, null) .asList().join().get(0).getPayload(); - assertThat(payload.hasOldRecords(), is(false)); - assertThat(payload.hasNewRecord(), is(true)); + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); + assertThat(records.hasOldRecords(), is(false)); + assertThat(records.hasNewRecord(), is(true)); final TestRecords1Proto.MySimpleRecord enqueued = TestRecords1Proto.MySimpleRecord.newBuilder() .mergeFrom(recordStore.getSerializer().deserialize(recordStore.getRecordMetaData(), - TupleHelpers.EMPTY, payload.getNewRecord().toByteArray(), recordStore.getTimer())) + TupleHelpers.EMPTY, records.getNewRecord().toByteArray(), recordStore.getTimer())) .build(); assertThat(enqueued.getRecNo(), is(1066L)); assertThat(enqueued.getNumValue3Indexed(), is(42)); diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java index 3ec09ce7d3..8211fac555 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java @@ -125,7 +125,7 @@ void testDrainPendingQueueWhileBuilding(int limit) throws Exception { .map(entry -> { final IndexBuildProto.PendingWritesQueueEntry payload = entry.getPayload(); final Message rec = recordStore.getSerializer().deserialize(recordStore.getRecordMetaData(), - TupleHelpers.EMPTY, payload.getNewRecord().toByteArray(), recordStore.getTimer()); + TupleHelpers.EMPTY, payload.getOldAndNewRecords().getNewRecord().toByteArray(), recordStore.getTimer()); return (Long)rec.getField(rec.getDescriptorForType().findFieldByName("rec_no")); }) .toList(); @@ -190,7 +190,7 @@ void testDrainPendingQueueWhileBuildingAfterNPasses(final int passesCount) throw .map(entry -> { final IndexBuildProto.PendingWritesQueueEntry payload = entry.getPayload(); final Message record = recordStore.getSerializer().deserialize(recordStore.getRecordMetaData(), - TupleHelpers.EMPTY, payload.getNewRecord().toByteArray(), recordStore.getTimer()); + TupleHelpers.EMPTY, payload.getOldAndNewRecords().getNewRecord().toByteArray(), recordStore.getTimer()); return (Long)record.getField(record.getDescriptorForType().findFieldByName("rec_no")); }) .toList(); From fcb278a72b835060684c299287ff705aad868c28 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 14:11:11 -0400 Subject: [PATCH 06/15] Move serialization to the index maintainers --- .../IndexingPendingWriteQueue.java | 38 +++-------------- .../foundationdb/IndexingThrottle.java | 11 +++-- .../indexes/SlidingWindowIndexMaintainer.java | 3 +- .../indexes/StandardIndexMaintainer.java | 42 ++++++++++++++++++- 4 files changed, 53 insertions(+), 41 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index 8db464fbde..ad3c010f79 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -130,48 +130,20 @@ public static PendingWritesQueue getInd } /** - * Called by the index maintainer to serialize an old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry}. - * @param store the record store whose serializer, incarnation, and context are used + * 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 oldRecord the record prior to the update, or null for an insert - * @param newRecord the record after the update, or null for a delete - * @param the message type of the records + * @param entry the entry to enqueue * @return a future that completes when the entry has been enqueued */ @Nonnull - public static CompletableFuture enqueueOldAndNewRecords( + public static CompletableFuture enqueuePendingIndexUpdate( final FDBRecordStore store, final Index index, - @Nullable final FDBIndexableRecord oldRecord, - @Nullable final FDBIndexableRecord newRecord) { - if (oldRecord == null && newRecord == null) { - // This should never happen with the current maintainers - return AsyncUtil.DONE; - } - final RecordSerializer serializer = store.getSerializer(); - final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords.Builder recordsBuilder = - IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords.newBuilder(); - if (oldRecord != null) { - recordsBuilder.setOldRecords(serializeRecord(store, oldRecord, serializer)); - } - if (newRecord != null) { - recordsBuilder.setNewRecord(serializeRecord(store, newRecord, serializer)); - } - final IndexBuildProto.PendingWritesQueueEntry entry = - IndexBuildProto.PendingWritesQueueEntry.newBuilder() - .setOldAndNewRecords(recordsBuilder) - .build(); + final IndexBuildProto.PendingWritesQueueEntry entry) { return getIndexingQueue(store, index).enqueue(store.getContext(), entry, store.getIncarnation()); } - @Nonnull - private static ByteString serializeRecord(final FDBRecordStore store, - final FDBIndexableRecord indexableRecord, - final RecordSerializer serializer) { - return ByteString.copyFrom(serializer.serialize(store.getRecordMetaData(), - indexableRecord.getRecordType(), indexableRecord.getRecord(), store.getTimer())); - } - @Nullable public static FDBStoredRecord getOldRecord(final FDBRecordStore store, IndexBuildProto.PendingWritesQueueEntry payload) { final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java index 1fa5aa2391..c452984217 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java @@ -468,12 +468,11 @@ private CompletableFuture populateDrainRequiredIndexes(FDBRecordStore stor */ private CompletableFuture> nonEmptyQueueIndexes(@Nonnull FDBRecordStore store, @Nonnull FDBRecordContext context, @Nonnull List indexes) { return AsyncUtil.getAll(indexes.stream() - .map(index -> { - final PendingWritesQueue queue = - IndexingPendingWriteQueue.getIndexingQueue(store, index); - return queue.getQueueSizeNoConflict(context) - .thenApply(size -> size != null && size > 0 ? index : null); - }) + .map(index -> + IndexingPendingWriteQueue.getIndexingQueue(store, index) + .getQueueSizeNoConflict(context) + .thenApply(size -> size != null && size > 0 ? index : null) + ) .toList() ) .thenApply(results -> results.stream().filter(Objects::nonNull).toList()); diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java index b640f8df3b..c56293b837 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java @@ -431,7 +431,8 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - return IndexingPendingWriteQueue.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); + return IndexingPendingWriteQueue.enqueuePendingIndexUpdate(state.store, state.index, + StandardIndexMaintainer.buildPendingWritesQueueEntry(state, oldRecord, newRecord)); } /** diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index 482fed562e..547fb08050 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -31,6 +31,7 @@ import com.apple.foundationdb.record.CursorStreamingMode; import com.apple.foundationdb.record.EvaluationContext; import com.apple.foundationdb.record.ExecuteProperties; +import com.apple.foundationdb.record.IndexBuildProto; import com.apple.foundationdb.record.IndexEntry; import com.apple.foundationdb.record.IndexScanType; import com.apple.foundationdb.record.IsolationLevel; @@ -51,6 +52,7 @@ import com.apple.foundationdb.record.metadata.expressions.GroupingKeyExpression; import com.apple.foundationdb.record.metadata.expressions.KeyExpression; import com.apple.foundationdb.record.metadata.expressions.KeyWithValueExpression; +import com.apple.foundationdb.record.provider.common.RecordSerializer; import com.apple.foundationdb.record.provider.foundationdb.FDBExceptions; import com.apple.foundationdb.record.provider.foundationdb.FDBIndexableRecord; import com.apple.foundationdb.record.provider.foundationdb.FDBIndexedRawRecord; @@ -74,6 +76,7 @@ import com.apple.foundationdb.tuple.Tuple; import com.apple.foundationdb.tuple.TupleHelpers; import com.google.common.base.Verify; +import com.google.protobuf.ByteString; import com.google.protobuf.Message; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -339,7 +342,44 @@ public CompletableFuture updateWhileWriteOnlyWithQueue // had generated an "already indexed" source index key, the other one should be converted to a null in the queue). // This, however can be done at a later step. } - return IndexingPendingWriteQueue.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); + return IndexingPendingWriteQueue.enqueuePendingIndexUpdate(state.store, state.index, + buildPendingWritesQueueEntry(state, oldRecord, newRecord)); + } + + /** + * Serialize an old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry} so that it can be deferred + * onto the index's pending writes queue by {@link #updateWhileWriteOnlyWithQueue}. Either record may be null: a null + * {@code oldRecord} represents an insert and a null {@code newRecord} represents a delete. + * @param state the maintainer state whose store serializer is used + * @param oldRecord the previous stored record or {@code null} if a new record is being created + * @param newRecord the new record or {@code null} if an old record is being deleted + * @param type of message + * @return the entry to enqueue + */ + @Nonnull + static IndexBuildProto.PendingWritesQueueEntry buildPendingWritesQueueEntry(@Nonnull final IndexMaintainerState state, + @Nullable final FDBIndexableRecord oldRecord, + @Nullable final FDBIndexableRecord newRecord) { + final RecordSerializer serializer = state.store.getSerializer(); + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords.Builder recordsBuilder = + IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords.newBuilder(); + if (oldRecord != null) { + recordsBuilder.setOldRecords(serializeRecord(state, oldRecord, serializer)); + } + if (newRecord != null) { + recordsBuilder.setNewRecord(serializeRecord(state, newRecord, serializer)); + } + return IndexBuildProto.PendingWritesQueueEntry.newBuilder() + .setOldAndNewRecords(recordsBuilder) + .build(); + } + + @Nonnull + private static ByteString serializeRecord(@Nonnull final IndexMaintainerState state, + @Nonnull final FDBIndexableRecord indexableRecord, + @Nonnull final RecordSerializer serializer) { + return ByteString.copyFrom(serializer.serialize(state.store.getRecordMetaData(), + indexableRecord.getRecordType(), indexableRecord.getRecord(), state.store.getTimer())); } private boolean isSynthetic() { From bc2c9de27c5d9a1ad489e3f574a2f092b905467f Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 14:28:26 -0400 Subject: [PATCH 07/15] Close SW test gap --- .../foundationdb/IndexingPendingWriteQueue.java | 15 +++++++++++++++ .../provider/foundationdb/IndexingThrottle.java | 5 ++--- .../indexes/SlidingWindowIndexTest.java | 6 +++++- 3 files changed, 22 insertions(+), 4 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index ad3c010f79..cecba9bc8b 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -35,6 +35,7 @@ import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.ThrottledRetryingIterator; import com.apple.foundationdb.tuple.TupleHelpers; import com.apple.foundationdb.util.CloseException; +import com.google.common.annotations.VisibleForTesting; import com.google.protobuf.ByteString; import com.google.protobuf.Message; @@ -129,6 +130,20 @@ public static PendingWritesQueue getInd ); } + /** + * 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 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 diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java index c452984217..c551b50aa3 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java @@ -469,9 +469,8 @@ private CompletableFuture populateDrainRequiredIndexes(FDBRecordStore stor private CompletableFuture> nonEmptyQueueIndexes(@Nonnull FDBRecordStore store, @Nonnull FDBRecordContext context, @Nonnull List indexes) { return AsyncUtil.getAll(indexes.stream() .map(index -> - IndexingPendingWriteQueue.getIndexingQueue(store, index) - .getQueueSizeNoConflict(context) - .thenApply(size -> size != null && size > 0 ? index : null) + IndexingPendingWriteQueue.hasPendingWrites(store, index, context) + .thenApply(hasPending -> hasPending ? index : null) ) .toList() ) diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java index ee11318507..3ab9d54ac5 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java @@ -1494,6 +1494,8 @@ private static class StubIndexMaintainer extends IndexMaintainer { new IndexEntry(null, Tuple.from(1L), Tuple.from()); private static final IndexOperationResult SENTINEL_OP_RESULT = new IndexOperationResult() { }; + private boolean pendingWriteQueueAllowed = true; + StubIndexMaintainer(@Nonnull IndexMaintainerState state) { super(state); } @@ -1593,7 +1595,7 @@ public boolean isIdempotent() { @Override public boolean isPendingWriteQueueAllowed() { - return true; + return pendingWriteQueueAllowed; } @Nonnull @@ -1701,6 +1703,8 @@ void delegateMethodsWithMock() throws Exception { // mergeIndex sw.mergeIndex().join(); + // isPendingWriteQueueAllowed (delegated to the wrapped maintainer) + assertTrue(sw.isPendingWriteQueueAllowed()); commit(context); } } From 154474d40bccc0aff6fd5bfb87ec09fb08d89567 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 14:52:35 -0400 Subject: [PATCH 08/15] Move queue entry handling from the drain to the index maintainers --- .../foundationdb/IndexMaintainer.java | 12 +++++ .../IndexingPendingWriteQueue.java | 40 +-------------- .../indexes/NoOpIndexMaintainer.java | 7 +++ .../indexes/SlidingWindowIndexMaintainer.java | 10 ++++ .../indexes/StandardIndexMaintainer.java | 50 +++++++++++++++++++ .../FDBRecordStoreUniqueIndexTest.java | 7 +++ .../foundationdb/OnlineIndexerMergeTest.java | 7 +++ .../foundationdb/TerribleIndexMaintainer.java | 7 +++ .../indexes/SlidingWindowIndexTest.java | 6 +++ 9 files changed, 107 insertions(+), 39 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java index ac87b755b6..9017f35c1c 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java +++ b/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; @@ -173,6 +174,17 @@ public abstract CompletableFuture updateWhileWriteOnly @Nullable FDBIndexableRecord 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 updateFromQueue(@Nonnull IndexBuildProto.PendingWritesQueueEntry payload); + + /** * Scans through the list of uniqueness violations within the database. *

diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index cecba9bc8b..08408e2c4b 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -24,23 +24,15 @@ import com.apple.foundationdb.record.IndexBuildProto; import com.apple.foundationdb.record.RecordCoreException; import com.apple.foundationdb.record.RecordCursorResult; -import com.apple.foundationdb.record.RecordMetaData; import com.apple.foundationdb.record.ScanProperties; import com.apple.foundationdb.record.metadata.Index; -import com.apple.foundationdb.record.metadata.RecordType; -import com.apple.foundationdb.record.provider.common.RecordSerializer; import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueue; import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueueEntry; import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.CursorFactory; import com.apple.foundationdb.record.provider.foundationdb.runners.throttled.ThrottledRetryingIterator; -import com.apple.foundationdb.tuple.TupleHelpers; import com.apple.foundationdb.util.CloseException; -import com.google.common.annotations.VisibleForTesting; -import com.google.protobuf.ByteString; -import com.google.protobuf.Message; import javax.annotation.Nonnull; -import javax.annotation.Nullable; import javax.annotation.ParametersAreNonnullByDefault; import java.io.Serial; import java.util.concurrent.CompletableFuture; @@ -110,10 +102,7 @@ private CompletableFuture 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( - getOldRecord(store, payload), - getNewRecord(store, payload)) + .updateFromQueue(payload) .thenAccept(ignore -> { quotaManager.deleteCountInc(); getIndexingQueue(store, index).clearEntry(store.getContext(), entry); @@ -159,33 +148,6 @@ public static CompletableFuture enqueuePendingIndexUpdate( return getIndexingQueue(store, index).enqueue(store.getContext(), entry, store.getIncarnation()); } - @Nullable - public static FDBStoredRecord getOldRecord(final FDBRecordStore store, IndexBuildProto.PendingWritesQueueEntry payload) { - final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); - return records.hasOldRecords() ? deserializeRecord(store, records.getOldRecords()) : null; - } - - @Nullable - public static FDBStoredRecord getNewRecord(final FDBRecordStore store, IndexBuildProto.PendingWritesQueueEntry payload) { - final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); - return records.hasNewRecord() ? deserializeRecord(store, records.getNewRecord()) : null; - } - - /** - * Rebuild a stored record from its serialized payload. The queue holds only the record bytes, so the primary key is - * recomputed from the record's metadata, mirroring {@link FDBRecordStore#saveTypedRecord}. - */ - @Nonnull - private static FDBStoredRecord deserializeRecord(final FDBRecordStore store, final ByteString serialized) { - final RecordMetaData metaData = store.getRecordMetaData(); - final RecordSerializer serializer = store.getSerializer(); - final Message rec = serializer.deserialize(metaData, TupleHelpers.EMPTY, serialized.toByteArray(), store.getTimer()); - final RecordType recordType = metaData.getRecordTypeForDescriptor(rec.getDescriptorForType()); - final FDBStoredRecordBuilder builder = FDBStoredRecord.newBuilder(rec).setRecordType(recordType); - builder.setPrimaryKey(recordType.getPrimaryKey().evaluateSingleton(builder).toTuple()); - return builder.build(); - } - /** * thrown if pending queue drain had failed. */ diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java index 8d1c6a49ae..9bea1d1f1b 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.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; @@ -86,6 +87,12 @@ public CompletableFuture updateWhileWriteOnlyWithQueue return AsyncUtil.DONE; } + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + return AsyncUtil.DONE; + } + @Nonnull @Override public RecordCursor scanUniquenessViolations(@Nonnull TupleRange range, @Nullable byte[] continuation, @Nonnull ScanProperties scanProperties) { diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java index c56293b837..173656574a 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java @@ -26,6 +26,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; @@ -435,6 +436,15 @@ public CompletableFuture updateWhileWriteOnlyWithQueue StandardIndexMaintainer.buildPendingWritesQueueEntry(state, oldRecord, newRecord)); } + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + // Apply via updateWhileWriteOnly (the sliding-window semantics), lest this update be re-pushed to the queue + return updateWhileWriteOnly( + StandardIndexMaintainer.getOldRecord(state, payload), + StandardIndexMaintainer.getNewRecord(state, payload)); + } + /** * Evaluates the partition key from a record. Returns an empty tuple if there is no partition. */ diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index 547fb08050..d58b9350da 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -40,6 +40,7 @@ import com.apple.foundationdb.record.RecordCoreException; import com.apple.foundationdb.record.RecordCursor; import com.apple.foundationdb.record.RecordIndexUniquenessViolation; +import com.apple.foundationdb.record.RecordMetaData; import com.apple.foundationdb.record.ScanProperties; import com.apple.foundationdb.record.TupleRange; import com.apple.foundationdb.record.logging.KeyValueLogMessage; @@ -59,6 +60,8 @@ import com.apple.foundationdb.record.provider.foundationdb.FDBRecord; import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase; import com.apple.foundationdb.record.provider.foundationdb.FDBStoreTimer; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoredRecord; +import com.apple.foundationdb.record.provider.foundationdb.FDBStoredRecordBuilder; import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainer; import com.apple.foundationdb.record.provider.foundationdb.IndexMaintainerState; import com.apple.foundationdb.record.provider.foundationdb.IndexMaintenanceFilter; @@ -346,6 +349,15 @@ public CompletableFuture updateWhileWriteOnlyWithQueue buildPendingWritesQueueEntry(state, oldRecord, newRecord)); } + @Override + @Nonnull + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + // Calling updateWhileWriteOnly explicitly, lest this update be re-pushed to the queue + return updateWhileWriteOnly( + getOldRecord(state, payload), + getNewRecord(state, payload)); + } + /** * Serialize an old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry} so that it can be deferred * onto the index's pending writes queue by {@link #updateWhileWriteOnlyWithQueue}. Either record may be null: a null @@ -382,6 +394,44 @@ private static ByteString serializeRecord(@Nonnull final Ind indexableRecord.getRecordType(), indexableRecord.getRecord(), state.store.getTimer())); } + /** + * Deserialize the old records from a pending write queued. + * @param state the maintainer state whose store serializer is used + * @param payload the queued entry to read + * @return the previous stored record, or {@code null} if none + */ + @Nullable + static FDBStoredRecord getOldRecord(@Nonnull final IndexMaintainerState state, + @Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); + return records.hasOldRecords() ? deserializeRecord(state, records.getOldRecords()) : null; + } + + /** + * Deserialize the new record from a pending write queued. + * @param state the maintainer state whose store serializer is used + * @param payload the queued entry to read + * @return the new stored record, or {@code null} if none + */ + @Nullable + static FDBStoredRecord getNewRecord(@Nonnull final IndexMaintainerState state, + @Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); + return records.hasNewRecord() ? deserializeRecord(state, records.getNewRecord()) : null; + } + + @Nonnull + private static FDBStoredRecord deserializeRecord(@Nonnull final IndexMaintainerState state, + @Nonnull final ByteString serialized) { + final RecordMetaData metaData = state.store.getRecordMetaData(); + final RecordSerializer serializer = state.store.getSerializer(); + final Message rec = serializer.deserialize(metaData, TupleHelpers.EMPTY, serialized.toByteArray(), state.store.getTimer()); + final RecordType recordType = metaData.getRecordTypeForDescriptor(rec.getDescriptorForType()); + final FDBStoredRecordBuilder builder = FDBStoredRecord.newBuilder(rec).setRecordType(recordType); + builder.setPrimaryKey(recordType.getPrimaryKey().evaluateSingleton(builder).toTuple()); + return builder.build(); + } + private boolean isSynthetic() { return !state.store.getRecordMetaData().getSyntheticRecordTypes().isEmpty() && state.store.getRecordMetaData().recordTypesForIndex(state.index).stream().anyMatch(RecordType::isSynthetic); diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java index 4a6f2a5476..5aa0a79314 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java @@ -24,6 +24,7 @@ import com.apple.foundationdb.Transaction; import com.apple.foundationdb.async.MoreAsyncUtil; 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.IndexState; @@ -856,6 +857,12 @@ public CompletableFuture updateWhileWriteOnlyWithQueue return underlying.updateWhileWriteOnlyWithQueue(oldRecord, newRecord); } + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + return underlying.updateFromQueue(payload); + } + @Nonnull @Override public RecordCursor scanUniquenessViolations(@Nonnull final TupleRange range, @Nullable final byte[] continuation, @Nonnull final ScanProperties scanProperties) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java index 3d35fb6d6e..e15cc3d9b1 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java @@ -25,6 +25,7 @@ import com.apple.foundationdb.Transaction; 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; @@ -433,6 +434,12 @@ public CompletableFuture updateWhileWriteOnlyWithQueue throw new UnsupportedOperationException(); } + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + throw new UnsupportedOperationException(); + } + @Nonnull @Override public RecordCursor scanUniquenessViolations(@Nonnull final TupleRange range, @Nullable final byte[] continuation, @Nonnull final ScanProperties scanProperties) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java index 8dfe6b01b3..2cae9b7de3 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java @@ -23,6 +23,7 @@ import com.apple.foundationdb.Transaction; 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; @@ -106,6 +107,12 @@ public CompletableFuture updateWhileWriteOnlyWithQueue return AsyncUtil.DONE; } + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + return AsyncUtil.DONE; + } + @Nonnull @Override public RecordCursor scanUniquenessViolations(@Nonnull TupleRange range, @Nullable byte[] continuation, @Nonnull ScanProperties scanProperties) { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java index 3ab9d54ac5..7337acd804 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java @@ -1528,6 +1528,12 @@ public CompletableFuture updateWhileWriteOnlyWithQueue throw new UnsupportedOperationException("sliding window should not delegate this call"); } + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + throw new UnsupportedOperationException("sliding window should not delegate this call"); + } + @Nonnull @Override public RecordCursor scanUniquenessViolations(@Nonnull TupleRange range, From ca05d9fc3b00d728cefe20a69056afa8055645bc Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 15:07:36 -0400 Subject: [PATCH 09/15] Remove unused imports --- .../record/provider/foundationdb/IndexingThrottle.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java index c551b50aa3..094e907508 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingThrottle.java @@ -24,7 +24,6 @@ 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; @@ -32,7 +31,6 @@ 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; From a57e6a002d4cceaa66166652f226f77e98014319 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 16:10:56 -0400 Subject: [PATCH 10/15] use proto "oneof" --- fdb-record-layer-core/src/main/proto/index_build.proto | 7 +++++-- .../provider/foundationdb/FDBRecordStoreIndexTest.java | 5 +++++ 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/fdb-record-layer-core/src/main/proto/index_build.proto b/fdb-record-layer-core/src/main/proto/index_build.proto index da09e01ece..967ff07dcf 100644 --- a/fdb-record-layer-core/src/main/proto/index_build.proto +++ b/fdb-record-layer-core/src/main/proto/index_build.proto @@ -57,8 +57,9 @@ message IndexBuildHeartbeat { } // The message that gets saved into the pending writes queue. +// This message gets stored in an google.protobuf.Any field in the queue. This means that this message type SHOULD NOT be moved +// or renamed to preserve wire-format compatibility with existing data. message PendingWritesQueueEntry { - option (com.apple.foundationdb.record.record).usage = UNION; message OldAndNewRecords { // Simple cases where store serializer can handle the records (typically this is not the case for synthetic records) @@ -66,5 +67,7 @@ message PendingWritesQueueEntry { optional bytes new_record = 2; } - optional OldAndNewRecords oldAndNewRecords = 1; + oneof entry { + OldAndNewRecords oldAndNewRecords = 1; + } } diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java index 2e068a5968..29030ab9d7 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java @@ -1027,6 +1027,11 @@ void writeOnlyWithQueueIndex() throws Exception { assertThat(IndexingPendingWriteQueue.getIndexingQueue(recordStore, permissiveIndex) .getQueueSizeNoConflict(context).join(), is(nullValue())); + // Draining a queued entry through the NoOp maintainer is also a no-op that completes without applying + // anything to the index (the permissive index never defers real work to the queue). + recordStore.getIndexMaintainer(permissiveIndex) + .updateFromQueue(IndexBuildProto.PendingWritesQueueEntry.getDefaultInstance()).join(); + commit(context); } } From 2c174de7ba5e49d09b93e3747fcd89582c096dce Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 16:17:14 -0400 Subject: [PATCH 11/15] Remove import --- fdb-record-layer-core/src/main/proto/index_build.proto | 2 -- 1 file changed, 2 deletions(-) diff --git a/fdb-record-layer-core/src/main/proto/index_build.proto b/fdb-record-layer-core/src/main/proto/index_build.proto index 967ff07dcf..cb4b4e35c8 100644 --- a/fdb-record-layer-core/src/main/proto/index_build.proto +++ b/fdb-record-layer-core/src/main/proto/index_build.proto @@ -23,8 +23,6 @@ syntax = "proto2"; package com.apple.foundationdb.record; option java_outer_classname = "IndexBuildProto"; -import "record_metadata_options.proto"; - // This stamp indicates an ongoing indexing process. It will be used to assure a coherent indexing // continuation - which continued by another process. From 1bfbb59790ab5b010c472334642a4caf2fe99669 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Tue, 14 Jul 2026 16:20:59 -0400 Subject: [PATCH 12/15] typo in javadoc --- .../record/provider/foundationdb/IndexingPendingWriteQueue.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index 08408e2c4b..04eabb06bc 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java @@ -39,7 +39,7 @@ /** * Use {@link PendingWritesQueue} to defer index updates while an index is being built. - * Indexin maintainer: will use this module to push items to the queue + * 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 From ca244be355d731a3f5c0fc76d9f8ecfc0e3642e7 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Wed, 15 Jul 2026 17:48:37 -0400 Subject: [PATCH 13/15] Implement requested changes --- .../foundationdb/indexes/StandardIndexMaintainer.java | 8 +++++++- .../foundationdb/indexes/SlidingWindowIndexTest.java | 4 +--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index d58b9350da..f593f883cd 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -337,7 +337,7 @@ private CompletableFuture updateWriteOnlyByIndex(@Nonn @Override @Nonnull public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - if (!isIdempotent()) { + if (!isPendingWriteQueueAllowed()) { // The indexer should have not allowed the WRITE_ONLY_WITH_QUEUE index state. This path should never be reached. throw new UnsupportedOperationException("write pending queue is not yet supported for non-idempotent indexes"); // To support non-idempotent index, we should use the same "is included in built range" condition as updateWhileWriteOnly and @@ -403,6 +403,9 @@ private static ByteString serializeRecord(@Nonnull final Ind @Nullable static FDBStoredRecord getOldRecord(@Nonnull final IndexMaintainerState state, @Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + if (!payload.hasOldAndNewRecords()) { + throw new RecordCoreException("wrong item in queue"); + } final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); return records.hasOldRecords() ? deserializeRecord(state, records.getOldRecords()) : null; } @@ -416,6 +419,9 @@ static FDBStoredRecord getOldRecord(@Nonnull final IndexMaintainerState @Nullable static FDBStoredRecord getNewRecord(@Nonnull final IndexMaintainerState state, @Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + if (!payload.hasOldAndNewRecords()) { + throw new RecordCoreException("wrong item in queue"); + } final IndexBuildProto.PendingWritesQueueEntry.OldAndNewRecords records = payload.getOldAndNewRecords(); return records.hasNewRecord() ? deserializeRecord(state, records.getNewRecord()) : null; } diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java index 7337acd804..620f794a1f 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java @@ -1494,8 +1494,6 @@ private static class StubIndexMaintainer extends IndexMaintainer { new IndexEntry(null, Tuple.from(1L), Tuple.from()); private static final IndexOperationResult SENTINEL_OP_RESULT = new IndexOperationResult() { }; - private boolean pendingWriteQueueAllowed = true; - StubIndexMaintainer(@Nonnull IndexMaintainerState state) { super(state); } @@ -1601,7 +1599,7 @@ public boolean isIdempotent() { @Override public boolean isPendingWriteQueueAllowed() { - return pendingWriteQueueAllowed; + return true; } @Nonnull From c2776019200fdbe4fd514946f68f0c0e914220a3 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Wed, 15 Jul 2026 18:35:06 -0400 Subject: [PATCH 14/15] Use serializePendingWriteQueue (instead of updateWhileWriteOnlyWithQueue) --- .../provider/foundationdb/FDBRecordStore.java | 8 ++++- .../foundationdb/IndexMaintainer.java | 23 ++++++-------- .../indexes/NoOpIndexMaintainer.java | 5 +-- .../indexes/SlidingWindowIndexMaintainer.java | 7 ++--- .../indexes/StandardIndexMaintainer.java | 19 +++--------- .../foundationdb/FDBRecordStoreIndexTest.java | 31 ++++++++++++------- .../FDBRecordStoreUniqueIndexTest.java | 4 +-- .../foundationdb/OnlineIndexerMergeTest.java | 2 +- .../foundationdb/TerribleIndexMaintainer.java | 4 +-- .../indexes/SlidingWindowIndexTest.java | 9 +++--- 10 files changed, 57 insertions(+), 55 deletions(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStore.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStore.java index 1aa10a4d20..eb819cd970 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStore.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStore.java @@ -779,7 +779,13 @@ private 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; diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java index 9017f35c1c..fda6474900 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java @@ -36,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; @@ -156,22 +155,21 @@ public abstract CompletableFuture 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. + *

+ * 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 null if a new record is being created * @param newRecord the new record or null if an old record is being deleted * @param 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 CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable FDBIndexableRecord oldRecord, - @Nullable FDBIndexableRecord newRecord); + public abstract IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable FDBIndexableRecord oldRecord, + @Nullable FDBIndexableRecord newRecord); /** @@ -331,8 +329,7 @@ protected CompletableFuture unsupportedAggregateFunction(@Nonnull IndexAg * {@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 {@link #updateWhileWriteOnlyWithQueue} cannot correctly defer - * and later replay updates should return {@code false} to refuse this state; + * 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(); diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java index 9bea1d1f1b..4bd24daeda 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/NoOpIndexMaintainer.java @@ -83,8 +83,9 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - return AsyncUtil.DONE; + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { + // isPendingWriteQueueAllowed() returns false, so this maintainer never defers writes to a queue. + throw new UnsupportedOperationException("NoOpIndexMaintainer does not support the pending write queue"); } @Nonnull diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java index 173656574a..72b7ef61be 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexMaintainer.java @@ -54,7 +54,6 @@ import com.apple.foundationdb.record.provider.foundationdb.IndexOperation; import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; import com.apple.foundationdb.record.logging.LogMessageKeys; -import com.apple.foundationdb.record.provider.foundationdb.IndexingPendingWriteQueue; import com.apple.foundationdb.record.query.QueryToKeyMatcher; import com.apple.foundationdb.subspace.Subspace; import com.apple.foundationdb.tuple.ByteArrayUtil; @@ -431,9 +430,9 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - return IndexingPendingWriteQueue.enqueuePendingIndexUpdate(state.store, state.index, - StandardIndexMaintainer.buildPendingWritesQueueEntry(state, oldRecord, newRecord)); + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { + // TODO: use delegate.serializePendingWriteQueue data and add it to a sliding window message. The correctly handle it in updateFromQueue + return StandardIndexMaintainer.buildPendingWritesQueueEntry(state, oldRecord, newRecord); } @Nonnull diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java index f593f883cd..4cb67d7cd8 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/indexes/StandardIndexMaintainer.java @@ -70,7 +70,6 @@ import com.apple.foundationdb.record.provider.foundationdb.IndexPrefetchRangeKeyValueCursor; import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; import com.apple.foundationdb.record.provider.foundationdb.IndexScanRange; -import com.apple.foundationdb.record.provider.foundationdb.IndexingPendingWriteQueue; import com.apple.foundationdb.record.provider.foundationdb.KeyValueCursor; import com.apple.foundationdb.record.provider.foundationdb.indexing.IndexingRangeSet; import com.apple.foundationdb.record.query.QueryToKeyMatcher; @@ -336,17 +335,9 @@ private CompletableFuture updateWriteOnlyByIndex(@Nonn @Override @Nonnull - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - if (!isPendingWriteQueueAllowed()) { - // The indexer should have not allowed the WRITE_ONLY_WITH_QUEUE index state. This path should never be reached. - throw new UnsupportedOperationException("write pending queue is not yet supported for non-idempotent indexes"); - // To support non-idempotent index, we should use the same "is included in built range" condition as updateWhileWriteOnly and - // only push to the queue items that were already indexed. (If being build by a source index, and only one of the old/new records - // had generated an "already indexed" source index key, the other one should be converted to a null in the queue). - // This, however can be done at a later step. - } - return IndexingPendingWriteQueue.enqueuePendingIndexUpdate(state.store, state.index, - buildPendingWritesQueueEntry(state, oldRecord, newRecord)); + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { + // TODO: forbid PWQ in the standard maintainer + return buildPendingWritesQueueEntry(state, oldRecord, newRecord); } @Override @@ -360,8 +351,8 @@ public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.Pe /** * Serialize an old/new record pair into a {@link IndexBuildProto.PendingWritesQueueEntry} so that it can be deferred - * onto the index's pending writes queue by {@link #updateWhileWriteOnlyWithQueue}. Either record may be null: a null - * {@code oldRecord} represents an insert and a null {@code newRecord} represents a delete. + * onto the index's pending writes queue. Either record may be null (but not both): a null {@code oldRecord} represents + * an insert and a null {@code newRecord} represents a delete. * @param state the maintainer state whose store serializer is used * @param oldRecord the previous stored record or {@code null} if a new record is being created * @param newRecord the new record or {@code null} if an old record is being deleted diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java index 29030ab9d7..1364ffc636 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreIndexTest.java @@ -132,7 +132,6 @@ import static org.hamcrest.Matchers.is; import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.hamcrest.Matchers.not; -import static org.hamcrest.Matchers.nullValue; import static org.hamcrest.Matchers.oneOf; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; @@ -959,9 +958,9 @@ void scanWriteOnlyIndex() throws Exception { void writeOnlyWithQueueIndex() throws Exception { final String standardIndexName = "MySimpleRecord$num_value_3_indexed"; final String permissiveIndexName = "permissive_index"; - // The "permissive" index type is backed by NoOpIndexMaintainer, while the value index above is - // backed by StandardIndexMaintainer. Marking both WRITE_ONLY_WITH_QUEUE and then saving a record - // exercises updateWhileWriteOnlyWithQueue on both maintainers. + // The "permissive" index type is backed by NoOpIndexMaintainer, which does not allow the pending write queue, + // while the value index above is backed by StandardIndexMaintainer, which does. Marking the standard index + // WRITE_ONLY_WITH_QUEUE and then saving a record exercises serializePendingWriteQueue on its maintainer. final RecordMetaDataHook hook = metaData -> metaData.addIndex("MySimpleRecord", new Index(permissiveIndexName, field("num_value_3_indexed"), "permissive")); @@ -981,7 +980,22 @@ void writeOnlyWithQueueIndex() throws Exception { assertThat(recordStore.isIndexWriteOnlyNoQueue(standardIndexName), is(false)); assertThat(recordStore.isIndexReadable(standardIndexName), is(false)); - // Saving a record routes the update through updateWhileWriteOnlyWithQueue for each maintainer + // The NoOp-backed permissive index does not allow the pending write queue, so saving a record while it is + // in WRITE_ONLY_WITH_QUEUE is rejected (the record store throws before anything is enqueued). + assertThrows(RecordCoreException.class, () -> recordStore.saveRecord(TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(1066L) + .setNumValue3Indexed(42) + .build())); + } + + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, hook); + final Index standardIndex = recordStore.getRecordMetaData().getIndex(standardIndexName); + + // Now only the standard index is deferred to the queue; the permissive index stays readable. + recordStore.markIndexWriteOnlyWithQueue(standardIndex).get(); + + // Saving a record routes the update through serializePendingWriteQueue for the standard maintainer // rather than writing the index entries directly. recordStore.saveRecord(TestRecords1Proto.MySimpleRecord.newBuilder() .setRecNo(1066L) @@ -996,7 +1010,6 @@ void writeOnlyWithQueueIndex() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context, hook); final Index standardIndex = recordStore.getRecordMetaData().getIndex(standardIndexName); - final Index permissiveIndex = recordStore.getRecordMetaData().getIndex(permissiveIndexName); // The StandardIndexMaintainer routed the update to its pending queue: exactly one deferred write. final PendingWritesQueue standardQueue = @@ -1022,13 +1035,9 @@ void writeOnlyWithQueueIndex() throws Exception { assertThat(context.ensureActive().getRange(recordStore.indexSubspace(standardIndex).range()).asList().join(), is(empty())); - // The NoOp-backed permissive index's updateWhileWriteOnlyWithQueue is a no-op, so nothing is enqueued - // for it (its queue size counter was never written). - assertThat(IndexingPendingWriteQueue.getIndexingQueue(recordStore, permissiveIndex) - .getQueueSizeNoConflict(context).join(), is(nullValue())); - // Draining a queued entry through the NoOp maintainer is also a no-op that completes without applying // anything to the index (the permissive index never defers real work to the queue). + final Index permissiveIndex = recordStore.getRecordMetaData().getIndex(permissiveIndexName); recordStore.getIndexMaintainer(permissiveIndex) .updateFromQueue(IndexBuildProto.PendingWritesQueueEntry.getDefaultInstance()).join(); diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java index 5aa0a79314..51c01f90dc 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreUniqueIndexTest.java @@ -853,8 +853,8 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - return underlying.updateWhileWriteOnlyWithQueue(oldRecord, newRecord); + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { + return underlying.serializePendingWriteQueue(oldRecord, newRecord); } @Nonnull diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java index e15cc3d9b1..56eebb5678 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerMergeTest.java @@ -430,7 +430,7 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { throw new UnsupportedOperationException(); } diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java index 2cae9b7de3..0380bef604 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java @@ -103,8 +103,8 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - return AsyncUtil.DONE; + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { + throw new UnsupportedOperationException("TerribleIndexMaintainer does not support the pending write queue"); } @Nonnull diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java index 620f794a1f..5116666e14 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowIndexTest.java @@ -1521,8 +1521,8 @@ public CompletableFuture updateWhileWriteOnly(@Nullabl @Nonnull @Override - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord o, - @Nullable final FDBIndexableRecord n) { + public IndexBuildProto.PendingWritesQueueEntry serializePendingWriteQueue(@Nullable final FDBIndexableRecord o, + @Nullable final FDBIndexableRecord n) { throw new UnsupportedOperationException("sliding window should not delegate this call"); } @@ -1715,9 +1715,8 @@ void delegateMethodsWithMock() throws Exception { @Test void writeOnlyWithQueueRoutesUpdatesToQueue() throws Exception { - // While the index is WRITE_ONLY_WITH_QUEUE, updates are routed to the pending queue via - // SlidingWindowIndexMaintainer.updateWhileWriteOnlyWithQueue instead of being written to the index. Once the - // indexer drains the queue, those updates are applied and the resulting window is valid. + // While the index is WRITE_ONLY_WITH_QUEUE, updates are routed to the pending queue instead of being written to + // the index. Once the indexer drains the queue, those updates are applied and the resulting window is valid. // Enqueue two updates while the index is in the queue state. try (FDBRecordContext context = openContext()) { openStore(context, 5, Direction.DESC); From ab53d1cbf3dae28fbcb2f8b8f4944eb2aa9aac32 Mon Sep 17 00:00:00 2001 From: Josef Ezra Date: Thu, 16 Jul 2026 11:20:13 -0400 Subject: [PATCH 15/15] Update fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java Co-authored-by: Scott Dugas --- .../record/provider/foundationdb/IndexMaintainer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java index fda6474900..97008a88e4 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexMaintainer.java @@ -329,7 +329,7 @@ protected CompletableFuture unsupportedAggregateFunction(@Nonnull IndexAg * {@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; + * Maintainers that 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();