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 5e8006a24e..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 @@ -24,6 +24,7 @@ import com.apple.foundationdb.annotation.API; import com.apple.foundationdb.async.AsyncUtil; import com.apple.foundationdb.record.EvaluationContext; +import com.apple.foundationdb.record.IndexBuildProto; import com.apple.foundationdb.record.IndexEntry; import com.apple.foundationdb.record.IndexScanType; import com.apple.foundationdb.record.IsolationLevel; @@ -35,7 +36,6 @@ import com.apple.foundationdb.record.metadata.IndexRecordFunction; import com.apple.foundationdb.record.metadata.Key; import com.apple.foundationdb.record.provider.foundationdb.indexes.InvalidIndexEntry; -import com.apple.foundationdb.record.provider.foundationdb.queue.PendingWritesQueue; import com.apple.foundationdb.record.query.QueryToKeyMatcher; import com.apple.foundationdb.subspace.Subspace; import com.apple.foundationdb.tuple.Tuple; @@ -155,22 +155,32 @@ public abstract 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); + + + /** + * 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); /** @@ -314,6 +324,16 @@ 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 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(); + /** * 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..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 @@ -283,13 +283,12 @@ 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.isPendingWriteQueueAllowed() && index.getRootExpression().versionColumns() == 0 && store.getFormatVersionEnum().isAtLeast(FormatVersion.WRITE_ONLY_WITH_QUEUE)) { - // TODO: support write-only-with-queue for synthetic records // TODO? support versioned index ("?" because these kind of indexes don't tend to have indexing bottlenecks) // TODO? support write-only-with-queue for mutual indexing (may require a drain semaphore) return store.markIndexWriteOnlyWithQueue(index); @@ -1079,11 +1078,11 @@ private synchronized IndexingMerger getIndexingMerger(Index index) { return indexingMergerMap.computeIfAbsent(index.getName(), k -> new IndexingMerger(index, common, policy.getInitialMergesCountLimit())); } - private synchronized PendingWriteQueueDrainer getIndexingDrainer(Index index) { + private synchronized IndexingPendingWriteQueue getIndexingDrainer(Index index) { if (indexingDrainerMap == null) { indexingDrainerMap = new HashMap<>(); } - return indexingDrainerMap.computeIfAbsent(index.getName(), k -> new PendingWriteQueueDrainer(index, common)); + return indexingDrainerMap.computeIfAbsent(index.getName(), k -> new IndexingPendingWriteQueue(index, common)); } private void deferAutoMergeDuringCommit(FDBRecordStore store) { 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/IndexingPendingWriteQueue.java similarity index 62% rename from fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueDrainer.java rename to fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingPendingWriteQueue.java index a92b419029..04eabb06bc 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/IndexingPendingWriteQueue.java @@ -1,9 +1,9 @@ /* - * PendingWriteQueueDrainer.java + * IndexingPendingWriteQueue.java * * This source file is part of the FoundationDB open source project * - * Copyright 2015-2023 Apple Inc. and the FoundationDB project authors + * Copyright 2015-2026 Apple Inc. and the FoundationDB project authors * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -38,28 +38,31 @@ import java.util.concurrent.CompletableFuture; /** - * Drain a pending writes queue. Used by the indexer. + * Use {@link PendingWritesQueue} to defer index updates while an index is being built. + * Indexing maintainer: will use this module to push items to the queue + * Online indexer: will use this module to drain the queue and update the index */ @ParametersAreNonnullByDefault -public class PendingWriteQueueDrainer { +public final class IndexingPendingWriteQueue { + // TODO: configurable maxQueueSize + private static final int MAX_QUEUE_SIZE = 100_000; + private static final int MAX_RECORDS_DELETE_PER_SECOND = 10_000; + private final Index index; private final IndexingCommon common; - private static final int MAX_RECORDS_DELETE_PER_SECOND = 10_000; - public PendingWriteQueueDrainer(final Index index, IndexingCommon common) { + public IndexingPendingWriteQueue(final Index index, final IndexingCommon common) { this.index = index; this.common = common; } CompletableFuture isQueueEmpty(FDBRecordStore store) { - return getQueue(store).isQueueEmpty(store.getContext()); + return getIndexingQueue(store, index).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. + // 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 = @@ -85,7 +88,7 @@ private CursorFactory { final byte[] continuation = lastResult == null ? null : lastResult.getContinuation().toBytes(); final ScanProperties scanProperties = ScanProperties.FORWARD_SCAN.with(props -> props.setReturnedRowLimit(rowLimit)); - return getQueue(store).getQueueCursor(store.getContext(), scanProperties, continuation); + return getIndexingQueue(store, index).getQueueCursor(store.getContext(), scanProperties, continuation); }; } @@ -99,19 +102,50 @@ private CompletableFuture handleOneItem(final FDBRecordStore store, } final IndexBuildProto.PendingWritesQueueEntry payload = entry.getPayload(); return store.getIndexMaintainer(index) - // Calling updateWhileWriteOnly explicitly, lest this update will be re-pushed to the queue - .updateWhileWriteOnly( - PendingWriteQueueIndexingFactory.getOldRecord(store, payload), - PendingWriteQueueIndexingFactory.getNewRecord(store, payload)) + .updateFromQueue(payload) .thenAccept(ignore -> { quotaManager.deleteCountInc(); - getQueue(store).clearEntry(store.getContext(), entry); + getIndexingQueue(store, index).clearEntry(store.getContext(), entry); }); } @Nonnull - private PendingWritesQueue getQueue(final FDBRecordStore store) { - return PendingWriteQueueIndexingFactory.getIndexingQueue(store, index); + public static PendingWritesQueue getIndexingQueue(final FDBRecordStore store, final Index index) { + return new PendingWritesQueue<>( + IndexingSubspaces.indexPendingWriteQueueSubspace(store, index), + IndexingSubspaces.indexPendingWriteQueueSizeSubspace(store, index), + MAX_QUEUE_SIZE, + IndexBuildProto.PendingWritesQueueEntry.class + ); + } + + /** + * Return true if the pending writes queue for the given index currently holds at least one entry. The size counter + * is read via a snapshot (conflict-free) read. + * @param store the record store whose queue is inspected + * @param index the index whose queue is inspected + * @param context the context used for the conflict-free read + * @return a future that completes with true if the queue is non-empty + */ + @Nonnull + public static CompletableFuture hasPendingWrites(final FDBRecordStore store, final Index index, final FDBRecordContext context) { + return getIndexingQueue(store, index).getQueueSizeNoConflict(context) + .thenApply(size -> size != null && size > 0); + } + + /** + * Called by the index maintainer to enqueue data for a deferred index update. + * @param store the record store whose incarnation and context are used + * @param index the index whose queue the entry is appended to + * @param entry the entry to enqueue + * @return a future that completes when the entry has been enqueued + */ + @Nonnull + public static CompletableFuture enqueuePendingIndexUpdate( + final FDBRecordStore store, + final Index index, + final IndexBuildProto.PendingWritesQueueEntry entry) { + return getIndexingQueue(store, index).enqueue(store.getContext(), entry, store.getIncarnation()); } /** 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..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; @@ -468,12 +466,10 @@ 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 = - PendingWriteQueueIndexingFactory.getIndexingQueue(store, index); - return queue.getQueueSizeNoConflict(context) - .thenApply(size -> size != null && size > 0 ? index : null); - }) + .map(index -> + IndexingPendingWriteQueue.hasPendingWrites(store, index, context) + .thenApply(hasPending -> hasPending ? index : null) + ) .toList() ) .thenApply(results -> results.stream().filter(Objects::nonNull).toList()); 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/PendingWriteQueueIndexingFactory.java deleted file mode 100644 index 2d0e14efb9..0000000000 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/PendingWriteQueueIndexingFactory.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * PendingWriteQueueIndexingFactory.java - * - * This source file is part of the FoundationDB open source project - * - * Copyright 2015-2026 Apple Inc. and the FoundationDB project authors - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * 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.RecordMetaData; -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.tuple.TupleHelpers; -import com.google.protobuf.ByteString; -import com.google.protobuf.Message; - -import javax.annotation.Nonnull; -import javax.annotation.Nullable; -import javax.annotation.ParametersAreNonnullByDefault; -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. - */ -@ParametersAreNonnullByDefault -public final class PendingWriteQueueIndexingFactory { - // TODO: configurable maxQueueSize - private static final int MAX_QUEUE_SIZE = 100_000; - - private PendingWriteQueueIndexingFactory() { - } - - @Nonnull - public static PendingWritesQueue getIndexingQueue(final FDBRecordStore store, final Index index) { - return new PendingWritesQueue<>( - IndexingSubspaces.indexPendingWriteQueueSubspace(store, index), - IndexingSubspaces.indexPendingWriteQueueSizeSubspace(store, index), - MAX_QUEUE_SIZE, - IndexBuildProto.PendingWritesQueueEntry.class - ); - } - - /** - * 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. - * @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 - * @param newRecord the record after the update, or null for a delete - * @param the message type of the records - * @return a future that completes when the entry has been enqueued - */ - @Nonnull - public static CompletableFuture enqueueOldAndNewRecords( - 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.Builder builder = IndexBuildProto.PendingWritesQueueEntry.newBuilder(); - if (oldRecord != null) { - builder.setOldRecords(serializeRecord(store, oldRecord, serializer)); - } - if (newRecord != null) { - builder.setNewRecord(serializeRecord(store, newRecord, serializer)); - } - return getIndexingQueue(store, index).enqueue(store.getContext(), builder.build(), 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) { - return payload.hasOldRecords() ? deserializeRecord(store, payload.getOldRecords()) : null; - } - - @Nullable - public static FDBStoredRecord getNewRecord(final FDBRecordStore store, IndexBuildProto.PendingWritesQueueEntry payload) { - return payload.hasNewRecord() ? deserializeRecord(store, payload.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(); - } - -} 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..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 @@ -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; @@ -82,7 +83,14 @@ 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) { + // isPendingWriteQueueAllowed() returns false, so this maintainer never defers writes to a queue. + throw new UnsupportedOperationException("NoOpIndexMaintainer does not support the pending write queue"); + } + + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { return AsyncUtil.DONE; } @@ -137,6 +145,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..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 @@ -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; @@ -53,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.PendingWriteQueueIndexingFactory; import com.apple.foundationdb.record.query.QueryToKeyMatcher; import com.apple.foundationdb.subspace.Subspace; import com.apple.foundationdb.tuple.ByteArrayUtil; @@ -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, @@ -425,8 +430,18 @@ 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); + 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 + @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)); } /** 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..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 @@ -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; @@ -39,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; @@ -51,12 +53,15 @@ 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; 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; @@ -66,7 +71,6 @@ import com.apple.foundationdb.record.provider.foundationdb.IndexScanBounds; import com.apple.foundationdb.record.provider.foundationdb.IndexScanRange; 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; @@ -74,6 +78,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; @@ -330,16 +335,103 @@ private CompletableFuture updateWriteOnlyByIndex(@Nonn @Override @Nonnull - public CompletableFuture updateWhileWriteOnlyWithQueue(@Nullable final FDBIndexableRecord oldRecord, @Nullable final FDBIndexableRecord newRecord) { - if (!isIdempotent()) { - // 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. + 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 + @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. 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 + * @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())); + } + + /** + * 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) { + if (!payload.hasOldAndNewRecords()) { + throw new RecordCoreException("wrong item in queue"); } - return PendingWriteQueueIndexingFactory.enqueueOldAndNewRecords(state.store, state.index, oldRecord, newRecord); + 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) { + 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; + } + + @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); } @Nullable @@ -750,6 +842,12 @@ public boolean isIdempotent() { return true; } + @Override + public boolean isPendingWriteQueueAllowed() { + // Synthetic records index maintainers that support pending write queues should override this function + return isIdempotent() && !isSynthetic(); + } + @Override @Nonnull public CompletableFuture addedRangeWithKey(@Nonnull Tuple primaryKey) { 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..cb4b4e35c8 100644 --- a/fdb-record-layer-core/src/main/proto/index_build.proto +++ b/fdb-record-layer-core/src/main/proto/index_build.proto @@ -56,11 +56,16 @@ 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. +// or renamed to preserve wire-format compatibility with existing data. message PendingWritesQueueEntry { - optional bytes old_records = 1; - optional bytes new_record = 2; + + 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; + } + + 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 f13d19573f..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,11 +1010,10 @@ 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 = - 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 @@ -1008,11 +1021,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)); @@ -1021,10 +1035,11 @@ 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(PendingWriteQueueIndexingFactory.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(); commit(context); } 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..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 @@ -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; @@ -852,8 +853,14 @@ 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 + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { + return underlying.updateFromQueue(payload); } @Nonnull @@ -912,6 +919,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..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 @@ -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; @@ -429,7 +430,13 @@ 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(); + } + + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { throw new UnsupportedOperationException(); } @@ -489,6 +496,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/OnlineIndexerPendingWriteQueueTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerPendingWriteQueueTest.java index 966edbed21..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 @@ -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); @@ -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(); @@ -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); @@ -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(); @@ -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()); } @@ -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/TerribleIndexMaintainer.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/TerribleIndexMaintainer.java index de736d8047..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 @@ -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; @@ -102,7 +103,13 @@ 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("TerribleIndexMaintainer does not support the pending write queue"); + } + + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { return AsyncUtil.DONE; } @@ -149,6 +156,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..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 @@ -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; @@ -1521,8 +1521,14 @@ 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"); + } + + @Nonnull + @Override + public CompletableFuture updateFromQueue(@Nonnull final IndexBuildProto.PendingWritesQueueEntry payload) { throw new UnsupportedOperationException("sliding window should not delegate this call"); } @@ -1591,6 +1597,11 @@ public boolean isIdempotent() { return true; } + @Override + public boolean isPendingWriteQueueAllowed() { + return true; + } + @Nonnull @Override public CompletableFuture addedRangeWithKey(@Nonnull Tuple primaryKey) { @@ -1696,15 +1707,16 @@ void delegateMethodsWithMock() throws Exception { // mergeIndex sw.mergeIndex().join(); + // isPendingWriteQueueAllowed (delegated to the wrapped maintainer) + assertTrue(sw.isPendingWriteQueueAllowed()); commit(context); } } @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); @@ -1825,7 +1837,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(); } }