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 cd88ee3b2b..d8d04d55ec 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 @@ -87,6 +87,9 @@ import com.apple.foundationdb.record.metadata.expressions.KeyExpression; import com.apple.foundationdb.record.provider.common.DynamicMessageRecordSerializer; import com.apple.foundationdb.record.provider.common.RecordSerializer; +import com.apple.foundationdb.record.provider.foundationdb.concurrency.FDBRecordStoreConcurrencyManager; +import com.apple.foundationdb.record.provider.foundationdb.concurrency.NoOpConcurrencyManager; +import com.apple.foundationdb.record.provider.foundationdb.concurrency.StoreConcurrencyManager; import com.apple.foundationdb.record.provider.foundationdb.indexing.IndexingHeartbeat; import com.apple.foundationdb.record.provider.foundationdb.indexing.IndexingRangeSet; import com.apple.foundationdb.record.provider.foundationdb.keyspace.KeySpacePath; @@ -302,6 +305,9 @@ public class FDBRecordStore extends FDBStoreBase implements FDBRecordStoreBase CompletableFuture> saveTypedRec final Tuple primaryKey = primaryKeyExpression.evaluateSingleton(recordBuilder).toTuple(); recordBuilder.setPrimaryKey(primaryKey); - final CompletableFuture> result = loadExistingRecord(typedSerializer, primaryKey).thenCompose(oldRecord -> { - if (oldRecord == null) { - if (existenceCheck.errorIfNotExists()) { - throw new RecordDoesNotExistException("record does not exist", - LogMessageKeys.PRIMARY_KEY, primaryKey); - } - } else { - if (existenceCheck.errorIfExists()) { - throw new RecordAlreadyExistsException("record already exists", - LogMessageKeys.PRIMARY_KEY, primaryKey); - } - if (existenceCheck.errorIfTypeChanged() && oldRecord.getRecordType() != recordType) { - throw new RecordTypeChangedException("record type changed", - LogMessageKeys.PRIMARY_KEY, primaryKey, - LogMessageKeys.ACTUAL_TYPE, oldRecord.getRecordType().getName(), - LogMessageKeys.EXPECTED_TYPE, recordType.getName()); - } - } - if (isDryRun) { - final FDBStoredRecord newRecord = dryRunSetSizeInfo(typedSerializer, recordBuilder, metaData); - return CompletableFuture.completedFuture(newRecord); - } - return getRecordStoreStateAsync().thenCompose(recordStoreState -> { - if (!overrideLock) { - validateRecordUpdateAllowed(recordStoreState); - } - final FDBStoredRecord newRecord = serializeAndSaveRecord(typedSerializer, recordBuilder, metaData, oldRecord); + return concurrencyManager.doWithRecordWriteLock(primaryKey, () -> { + final CompletableFuture> result = loadRecordForUpdate(typedSerializer, primaryKey).thenCompose(oldRecord -> { if (oldRecord == null) { - addRecordCount(metaData, newRecord, LITTLE_ENDIAN_INT64_ONE); + if (existenceCheck.errorIfNotExists()) { + throw new RecordDoesNotExistException("record does not exist", + LogMessageKeys.PRIMARY_KEY, primaryKey); + } } else { - if (getTimer() != null) { - getTimer().increment(FDBStoreTimer.Counts.REPLACE_RECORD_VALUE_BYTES, oldRecord.getValueSize()); + if (existenceCheck.errorIfExists()) { + throw new RecordAlreadyExistsException("record already exists", + LogMessageKeys.PRIMARY_KEY, primaryKey); + } + if (existenceCheck.errorIfTypeChanged() && oldRecord.getRecordType() != recordType) { + throw new RecordTypeChangedException("record type changed", + LogMessageKeys.PRIMARY_KEY, primaryKey, + LogMessageKeys.ACTUAL_TYPE, oldRecord.getRecordType().getName(), + LogMessageKeys.EXPECTED_TYPE, recordType.getName()); } } - return updateSecondaryIndexes(oldRecord, newRecord).thenApply(v -> newRecord); + if (isDryRun) { + final FDBStoredRecord newRecord = dryRunSetSizeInfo(typedSerializer, recordBuilder, metaData); + return CompletableFuture.completedFuture(newRecord); + } + return getRecordStoreStateAsync().thenCompose(recordStoreState -> { + if (!overrideLock) { + validateRecordUpdateAllowed(recordStoreState); + } + final FDBStoredRecord newRecord = serializeAndSaveRecord(typedSerializer, recordBuilder, metaData, oldRecord); + if (oldRecord == null) { + addRecordCount(metaData, newRecord, LITTLE_ENDIAN_INT64_ONE); + } else { + if (getTimer() != null) { + getTimer().increment(FDBStoreTimer.Counts.REPLACE_RECORD_VALUE_BYTES, oldRecord.getValueSize()); + } + } + return updateSecondaryIndexes(oldRecord, newRecord).thenApply(v -> newRecord); + }); }); + return context.instrument(FDBStoreTimer.Events.SAVE_RECORD, result); }); - return context.instrument(FDBStoreTimer.Events.SAVE_RECORD, result); } @SuppressWarnings("PMD.CloseResource") @@ -633,12 +643,16 @@ private FDBRecordVersion recordVersionForSave(@Nonnull RecordMetaData metaData, } @Nonnull - private CompletableFuture> loadExistingRecord(@Nonnull RecordSerializer typedSerializer, @Nonnull Tuple primaryKey) { + private CompletableFuture> loadRecordForUpdate(@Nonnull RecordSerializer typedSerializer, @Nonnull Tuple primaryKey) { // Note: this assumes that any existing record is compatible with the serializer (even if not of the same record type). // To relax that would perhaps mean catching errors and falling back to the untyped serializer. // This would in turn require care with the type parameters to updateSecondaryIndexes. // In no case is an index maintainer called with incompatible record type, so its signature should still be valid. - return loadTypedRecord(typedSerializer, primaryKey, false); + // + // This also calls the "Impl" variant directly, which means we skip grabbing an AsyncLock. This is deliberate, + // as this is designed for calls that take place within save and delete. Those methods already grab write locks + // over the record, and so attempting to grab a second lock would cause a deadlock (as the locks are not re-entrant). + return loadTypedRecordImpl(typedSerializer, primaryKey, ExecuteState.NO_LIMITS, false); } @Nonnull @@ -1057,6 +1071,15 @@ protected CompletableFuture> loadTypedRec @Nonnull final Tuple primaryKey, @Nonnull ExecuteState executeState, final boolean snapshot) { + return concurrencyManager.doWithRecordReadLock(primaryKey, + () -> loadTypedRecordImpl(typedSerializer, primaryKey, executeState, snapshot)); + } + + @Nonnull + private CompletableFuture> loadTypedRecordImpl(@Nonnull RecordSerializer typedSerializer, + @Nonnull final Tuple primaryKey, + @Nonnull ExecuteState executeState, + final boolean snapshot) { final RecordMetaData metaData = metaDataProvider.getRecordMetaData(); final Optional> versionFutureOptional; @@ -1074,7 +1097,7 @@ protected CompletableFuture> loadTypedRec byteScanLimiter.registerScannedBytes(sizeInfo.getKeySize() + sizeInfo.getValueSize()); } return rawRecord == null ? CompletableFuture.completedFuture(null) : - deserializeRecord(typedSerializer, rawRecord, metaData, versionFutureOptional); + deserializeRecord(typedSerializer, rawRecord, metaData, versionFutureOptional); }); return context.instrument(FDBStoreTimer.Events.LOAD_RECORD, result); } @@ -1238,15 +1261,16 @@ public void countKeyValue(@Nonnull final FDBStoreTimer.Count key, @Nonnull public CompletableFuture preloadRecordAsync(@Nonnull final Tuple primaryKey) { FDBPreloadRecordCache.Future futureRecord = preloadCache.beginPrefetch(primaryKey); - return loadRawRecordAsync(primaryKey, null, false) - .whenComplete((rawRecord, ex) -> { - if (ex != null) { - futureRecord.cancel(); - } else { - futureRecord.complete(rawRecord); - } - }) - .thenApply(rawRecord -> null); + return concurrencyManager.doWithRecordReadLock(primaryKey, () -> + loadRawRecordAsync(primaryKey, null, false) + .whenComplete((rawRecord, ex) -> { + if (ex != null) { + futureRecord.cancel(); + } else { + futureRecord.complete(rawRecord); + } + }) + ).thenApply(ignore -> null); } @Override @@ -1254,8 +1278,8 @@ public CompletableFuture preloadRecordAsync(@Nonnull final Tuple primaryKe public CompletableFuture recordExistsAsync(@Nonnull final Tuple primaryKey, @Nonnull final IsolationLevel isolationLevel) { final RecordMetaData metaData = metaDataProvider.getRecordMetaData(); final ReadTransaction tr = isolationLevel.isSnapshot() ? ensureContextActive().snapshot() : ensureContextActive(); - return SplitHelper.keyExists(tr, context, recordsSubspace(), - primaryKey, metaData.isSplitLongRecords(), omitUnsplitRecordSuffix); + return concurrencyManager.doWithRecordReadLock(primaryKey, + () -> SplitHelper.keyExists(tr, context, recordsSubspace(), primaryKey, metaData.isSplitLongRecords(), omitUnsplitRecordSuffix)); } @Nonnull @@ -1758,12 +1782,20 @@ public CompletableFuture deleteRecordAsync(@Nonnull final Tuple primary @Nonnull protected CompletableFuture deleteTypedRecord(@Nonnull RecordSerializer typedSerializer, @Nonnull Tuple primaryKey, boolean isDryRun) { + return concurrencyManager.doWithRecordWriteLock(primaryKey, + () -> deleteTypedRecordImpl(typedSerializer, primaryKey, isDryRun)); + } + + @Nonnull + private CompletableFuture deleteTypedRecordImpl(@Nonnull RecordSerializer typedSerializer, + @Nonnull Tuple primaryKey, boolean isDryRun) { if (isDryRun) { - return loadTypedRecord(typedSerializer, primaryKey, false).thenCompose(oldRecord -> oldRecord == null ? AsyncUtil.READY_FALSE : AsyncUtil.READY_TRUE); + return loadRecordForUpdate(typedSerializer, primaryKey) + .thenCompose(oldRecord -> oldRecord == null ? AsyncUtil.READY_FALSE : AsyncUtil.READY_TRUE); } preloadCache.invalidate(primaryKey); final RecordMetaData metaData = metaDataProvider.getRecordMetaData(); - CompletableFuture result = loadTypedRecord(typedSerializer, primaryKey, false).thenCompose(oldRecord -> { + CompletableFuture result = loadRecordForUpdate(typedSerializer, primaryKey).thenCompose(oldRecord -> { if (oldRecord == null) { return AsyncUtil.READY_FALSE; } @@ -5705,6 +5737,8 @@ public static class Builder implements BaseBuilder { @Nonnull private PlanSerializationRegistry planSerializationRegistry = DefaultPlanSerializationRegistry.INSTANCE; + private boolean disableConcurrencyManagement; + protected Builder() { } @@ -5735,6 +5769,7 @@ public final void copyFrom(@Nonnull Builder other) { this.stateCacheabilityOnOpen = other.stateCacheabilityOnOpen; this.bypassFullStoreLockReason = other.bypassFullStoreLockReason; this.planSerializationRegistry = other.planSerializationRegistry; + this.disableConcurrencyManagement = other.disableConcurrencyManagement; } /** @@ -5754,6 +5789,7 @@ public final void copyFrom(@Nonnull FDBRecordStore store) { this.storeStateCache = store.storeStateCache; this.stateCacheabilityOnOpen = store.stateCacheabilityOnOpen; this.planSerializationRegistry = store.planSerializationRegistry; + this.disableConcurrencyManagement = store.concurrencyManager instanceof NoOpConcurrencyManager; } @Override @@ -5968,6 +6004,18 @@ public void setPlanSerializationRegistry(@Nonnull final PlanSerializationRegistr this.planSerializationRegistry = planSerializationRegistry; } + @Override + public boolean isConcurrencyManagementDisabled() { + return disableConcurrencyManagement; + } + + @Override + @Nonnull + public Builder setDisableConcurrencyManagement(boolean disableConcurrencyManagement) { + this.disableConcurrencyManagement = disableConcurrencyManagement; + return this; + } + @Override @Nonnull public Builder copyBuilder() { @@ -5990,7 +6038,8 @@ public FDBRecordStore build() { } return new FDBRecordStore(context, subspaceProvider, formatVersion, getMetaDataProviderForBuild(), serializer, indexMaintainerRegistry, indexMaintenanceFilter, pipelineSizer, storeStateCache, stateCacheabilityOnOpen, - userVersionChecker, bypassFullStoreLockReason, planSerializationRegistry); + userVersionChecker, bypassFullStoreLockReason, planSerializationRegistry, + disableConcurrencyManagement ? NoOpConcurrencyManager.instance() : new FDBRecordStoreConcurrencyManager(subspaceProvider, context)); } @Override diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreBase.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreBase.java index d46b67f392..cc5b276a08 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreBase.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreBase.java @@ -71,6 +71,7 @@ import com.apple.foundationdb.subspace.Subspace; import com.apple.foundationdb.tuple.Tuple; import com.apple.foundationdb.tuple.TupleHelpers; +import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.protobuf.Message; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -2488,6 +2489,36 @@ default BaseBuilder setFormatVersion(FormatVersion formatVersion) { @Nonnull BaseBuilder setStateCacheabilityOnOpen(@Nonnull FDBRecordStore.StateCacheabilityOnOpen stateCacheabilityOnOpen); + /** + * Whether concurrency management is disabled. See {@link #setDisableConcurrencyManagement(boolean)} for + * more details. + * + * @return whether concurrency management is disabled + * @see com.apple.foundationdb.record.provider.foundationdb.concurrency.StoreConcurrencyManager + * @see #setDisableConcurrencyManagement(boolean) + */ + @API(API.Status.EXPERIMENTAL) + boolean isConcurrencyManagementDisabled(); + + /** + * Set whether to disable concurrency management. By default, the record store will use an + * {@link com.apple.foundationdb.record.provider.foundationdb.concurrency.StoreConcurrencyManager} + * to manage concurrent operations on the store. For example, the concurrency manager makes sure that + * concurrent operations that save the same record (in the same transaction) do not interfere with each + * other. If this is disabled, then those guardrails are removed, and it is the caller's responsibility + * to ensure that conflicting operations are not executed at the same time. In general, it is not + * recommended to run with this mode. The user should only opt in to this if they notice problems + * (e.g., the lock manager is too slow or something about their workflow causes a deadlock). + * + * @param disableConcurrencyManagement whether to disable concurrency management + * @return this builder + * @see com.apple.foundationdb.record.provider.foundationdb.concurrency.StoreConcurrencyManager + */ + @API(API.Status.EXPERIMENTAL) + @CanIgnoreReturnValue + @Nonnull + BaseBuilder setDisableConcurrencyManagement(boolean disableConcurrencyManagement); + /** * Make a copy of this builder. * This can be used to share enough of the state to connect to the same record store several times in different transactions. diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStore.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStore.java index 3a33ef4ab9..bd7398133e 100644 --- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStore.java +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStore.java @@ -578,6 +578,18 @@ public BaseBuilder> setStateCacheabilityOnOpen(@Nonnul return this; } + @Override + public boolean isConcurrencyManagementDisabled() { + return untypedStoreBuilder.isConcurrencyManagementDisabled(); + } + + @Nonnull + @Override + public Builder setDisableConcurrencyManagement(final boolean disableConcurrencyManagement) { + untypedStoreBuilder.setDisableConcurrencyManagement(disableConcurrencyManagement); + return this; + } + @Nullable @Override public String getBypassFullStoreLockReason() { diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/FDBRecordStoreConcurrencyManager.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/FDBRecordStoreConcurrencyManager.java new file mode 100644 index 0000000000..ccfca8e022 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/FDBRecordStoreConcurrencyManager.java @@ -0,0 +1,89 @@ +/* + * FDBRecordStoreConcurrencyManager.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.concurrency; + +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.record.locking.LockIdentifier; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordContext; +import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreKeyspace; +import com.apple.foundationdb.record.provider.foundationdb.SubspaceProvider; +import com.apple.foundationdb.subspace.Subspace; +import com.apple.foundationdb.tuple.Tuple; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; + +/** + * Default implementation of the {@link StoreConcurrencyManager}. It manages the locks used for + * read and write operations on behalf of the store. + */ +@API(API.Status.INTERNAL) +public final class FDBRecordStoreConcurrencyManager implements StoreConcurrencyManager { + @Nonnull + private final SubspaceProvider subspaceProvider; + @Nonnull + private final FDBRecordContext context; + @Nullable + private volatile Subspace recordSubspace; + + public FDBRecordStoreConcurrencyManager(@Nonnull SubspaceProvider subspaceProvider, @Nonnull FDBRecordContext context) { + this.subspaceProvider = subspaceProvider; + this.context = context; + } + + @Nonnull + private CompletableFuture getRecordsSubspaceAsync() { + Subspace cached = recordSubspace; + if (cached == null) { + // If we don't have a cached subspace, re-create it. We prefer this over using Suppliers::memoize + // so that subspace resolution throws a transient error, we don't memoize that future. + // It is possible that there are multiple subspace resolutions happening at the same time, but that's + // fine as they will all compute equivalent values, so it doesn't matter which one(s) win the race + // to set the cached subspace + return subspaceProvider.getSubspaceAsync(context).thenApply(storeSubspace -> { + final Subspace newRecordsSubspace = storeSubspace.subspace(Tuple.from(FDBRecordStoreKeyspace.RECORD.key())); + recordSubspace = newRecordsSubspace; + return newRecordsSubspace; + }); + } else { + return CompletableFuture.completedFuture(cached); + } + } + + @Nonnull + private CompletableFuture lockIdentifierForRecord(@Nonnull Tuple primaryKey) { + return getRecordsSubspaceAsync() + .thenApply(recordsSubspace -> recordsSubspace.subspace(primaryKey)) + .thenApply(LockIdentifier::new); + } + + @Override + public CompletableFuture doWithRecordReadLock(@Nonnull final Tuple primaryKey, @Nonnull final Supplier> operation) { + return lockIdentifierForRecord(primaryKey).thenCompose(id -> context.doWithReadLock(id, operation)); + } + + @Override + public CompletableFuture doWithRecordWriteLock(@Nonnull final Tuple primaryKey, @Nonnull final Supplier> operation) { + return lockIdentifierForRecord(primaryKey).thenCompose(id -> context.doWithWriteLock(id, operation)); + } +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/NoOpConcurrencyManager.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/NoOpConcurrencyManager.java new file mode 100644 index 0000000000..4eca510946 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/NoOpConcurrencyManager.java @@ -0,0 +1,63 @@ +/* + * NoOpConcurrencyManager.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.concurrency; + +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.tuple.Tuple; + +import javax.annotation.Nonnull; +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; + +/** + * Implementation of the {@link StoreConcurrencyManager} that does nothing. All operations are + * immediately allowed with no actual management of the concurrency occurring. + */ +@API(API.Status.INTERNAL) +public final class NoOpConcurrencyManager implements StoreConcurrencyManager { + @Nonnull + private static final NoOpConcurrencyManager INSTANCE = new NoOpConcurrencyManager(); + + private NoOpConcurrencyManager() { + // Singleton + } + + @Override + public CompletableFuture doWithRecordReadLock(@Nonnull final Tuple primaryKey, @Nonnull final Supplier> operation) { + return operation.get(); + } + + @Override + public CompletableFuture doWithRecordWriteLock(@Nonnull final Tuple primaryKey, @Nonnull final Supplier> operation) { + return operation.get(); + } + + /** + * Retrieve the instance for this class. The {@link NoOpConcurrencyManager} is a singleton, + * so this returns the same instance every time. + * + * @return the singleton instance + */ + @Nonnull + public static NoOpConcurrencyManager instance() { + return INSTANCE; + } +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/StoreConcurrencyManager.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/StoreConcurrencyManager.java new file mode 100644 index 0000000000..b4e01c8f92 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/StoreConcurrencyManager.java @@ -0,0 +1,90 @@ +/* + * FDBRecordStoreConcurrencyManager.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.concurrency; + +import com.apple.foundationdb.annotation.API; +import com.apple.foundationdb.record.query.expressions.QueryComponent; +import com.apple.foundationdb.tuple.Tuple; +import com.google.protobuf.Message; + +import javax.annotation.Nonnull; +import java.util.concurrent.CompletableFuture; +import java.util.function.Supplier; + +/** + * Interface for managing the concurrency within a given {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore}. + * That class is created with an instance of this interface as a member, and it should route operations through + * appropriate methods. + * + *

+ * This framework is a bit of a work in progress. It currently only offers basic protection for single record + * operations. That is, it prevents multiple operations from hitting issues when interleaving reads or writes + * to the same record, but it does not take any locks to prevent a concurrent read and a range operation like + * a {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore#deleteRecordsWhere(QueryComponent) deleteRecordsWhere()}. + * Nor will it prevent concurrent writes to the database during a query. As the framework gets refined, this + * may change. As such, the adopter is still somewhat responsible for managing their concurrent accesses to + * the record store until some of these shortcomings are addressed. + *

+ * + *

+ * If the lock management causes problems (e.g., if managing the locks takes too many resources or if the + * introduction of the lock manager results in deadlocks), this can be disabled by invoking + * {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase.BaseBuilder#setDisableConcurrencyManagement(boolean) setDisableConcurrencyManagement(true)} + * on the store's builder. This will switch the implementation to the {@link NoOpConcurrencyManager}, which + * runs all operations immediately without waiting for any lock. + *

+ * + * @see com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase.BaseBuilder#setDisableConcurrencyManagement(boolean) + * @see FDBRecordStoreConcurrencyManager for the default implementation + * @see NoOpConcurrencyManager for an implementation that does nothing, allowing the user to opt-out if problems arise + */ +@API(API.Status.INTERNAL) +public sealed interface StoreConcurrencyManager permits NoOpConcurrencyManager, FDBRecordStoreConcurrencyManager { + /** + * Perform an operation with a shared lock covering a single record. + * This is applied on operations like {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore#loadRecordAsync(Tuple) loadRecordAsync()} + * to ensure that the read does not see partial updates, e.g., one split point overwritten by a + * concurrent {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore#saveRecordAsync(Message) saveRecordAsync()}. + * This will wait for any previously started writes to the record to finish before beginning the operation, and it will + * block any future writes to the record from beginning until this read has completed. + * + * @param primaryKey the primary key of the record being read + * @param operation an operation to execute + * @return a future that will complete when the operation has finished + * @param the type returned by the operation + */ + CompletableFuture doWithRecordReadLock(@Nonnull Tuple primaryKey, @Nonnull Supplier> operation); + + /** + * Perform an operation with an exclusive lock covering a single record. + * This is applied on operations like {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore#saveRecordAsync(Message) saveRecordAsync()} + * and {@link com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore#deleteRecordAsync(Tuple) deleteRecordAsync()} + * to ensure that the writes do not interfere with each other or with any concurrent reads. + * This will wait for any previously started operations to the record to finish before beginning, and it will + * block any future operations to the record from beginning until this write has completed. + * + * @param primaryKey the primary key of the record being written + * @param operation an operation to execute + * @return a future that will complete when the operation has finished + * @param the type returned by the operation + */ + CompletableFuture doWithRecordWriteLock(@Nonnull Tuple primaryKey, @Nonnull Supplier> operation); +} diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/package-info.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/package-info.java new file mode 100644 index 0000000000..a3e57d0cc2 --- /dev/null +++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/concurrency/package-info.java @@ -0,0 +1,37 @@ +/* + * package-info.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. + */ + +/** + * Classes for managing concurrency of a store within a single transaction. For managing + * concurrency across different transactions, we generally rely on the FDB transaction + * resolver. That will fail any transaction that relies on stale information. However, + * within a transaction, the FDB client itself offers little help, and so locks need + * to be managed by the Record Layer framework itself. We also need to avoid taking normal + * locks over long-running operations, as that can block threads in the asynchronous thread pool. + * For that reason, we prefer using the {@link com.apple.foundationdb.record.locking.AsyncLock} + * abstraction, which returns a future that is made available only when a resource is + * ready. Each async lock is associated with a given resource key (that is, a + * {@link com.apple.foundationdb.record.locking.LockIdentifier}), and so the classes in + * this package are primarily present to ensure that we manage those locks the right way. + * + * @see com.apple.foundationdb.record.locking.AsyncLock + * @see com.apple.foundationdb.record.locking.LockRegistry + */ +package com.apple.foundationdb.record.provider.foundationdb.concurrency; diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreCrudTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreCrudTest.java index 677d62629c..d416af9627 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreCrudTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreCrudTest.java @@ -22,42 +22,102 @@ import com.apple.foundationdb.FDBError; import com.apple.foundationdb.FDBException; +import com.apple.foundationdb.async.AsyncUtil; import com.apple.foundationdb.record.IsolationLevel; import com.apple.foundationdb.record.RecordMetaData; import com.apple.foundationdb.record.TestRecords1Proto; import com.apple.foundationdb.record.TestRecordsBytesProto; import com.apple.foundationdb.record.TestRecordsUuidProto; import com.apple.foundationdb.record.TestRecordsWithUnionProto; +import com.apple.foundationdb.record.metadata.Index; import com.apple.foundationdb.record.metadata.MetaDataException; import com.apple.foundationdb.record.metadata.expressions.TupleFieldsHelper; +import com.apple.foundationdb.record.util.pair.Pair; import com.apple.foundationdb.tuple.Tuple; +import com.apple.test.BooleanSource; +import com.apple.test.RandomSeedSource; import com.apple.test.Tags; +import com.google.common.base.Strings; import com.google.protobuf.Message; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.parallel.Execution; import org.junit.jupiter.api.parallel.ExecutionMode; - +import org.junit.jupiter.params.ParameterizedTest; + +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Queue; +import java.util.Random; +import java.util.Set; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import java.util.stream.Stream; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.in; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.core.Is.is; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; /** * Basic CRUD operation tests on {@link FDBRecordStore}. */ @Tag(Tags.RequiresFDB) @Execution(ExecutionMode.CONCURRENT) -public class FDBRecordStoreCrudTest extends FDBRecordStoreTestBase { +class FDBRecordStoreCrudTest extends FDBRecordStoreTestBase { + @Nonnull + private final String longString = Strings.repeat("x", 101_000); + + /** + * Helper method to run index scrubbing validation on all indexes. This method + * assumes that the store has already been opened (see, e.g., {@link #openSimpleRecordStore(FDBRecordContext)}). + * It will ignore any scrubbing failures that happen because the index does not + * support scrubbing, but will assert that all other scrubbing jobs find no + * inconsistencies. + */ + private void scrubAllIndexes() { + for (Index index : recordStore.getRecordMetaData().getAllIndexes()) { + try (OnlineIndexScrubber scrubber = OnlineIndexScrubber.newBuilder() + .setRecordStore(recordStore) + .setIndex(index) + .setScrubbingPolicy(OnlineIndexScrubber.ScrubbingPolicy.newBuilder() + .setAllowRepair(false) + ) + .build()) { + assertEquals(0L, scrubber.scrubDanglingIndexEntries()); + assertEquals(0L, scrubber.scrubMissingIndexEntries()); + } catch (UnsupportedOperationException e) { + // Not all indexes support scrubbing. Ignore the ones where this fails + if (e.getMessage().contains("This index does not support scrubbing")) { + continue; + } + throw e; + } + } + } @Test - public void writeRead() throws Exception { + void writeRead() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -81,7 +141,7 @@ public void writeRead() throws Exception { } @Test - public void writeCheckExists() throws Exception { + void writeCheckExists() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -103,7 +163,7 @@ public void writeCheckExists() throws Exception { } @Test - public void writeCheckExistsConcurrently() throws Exception { + void writeCheckExistsConcurrently() throws Exception { try (FDBRecordContext context1 = openContext(); FDBRecordContext context2 = openContext()) { openSimpleRecordStore(context1); @@ -140,12 +200,13 @@ public void writeCheckExistsConcurrently() throws Exception { openSimpleRecordStore(context); assertThat(recordStore.recordExists(Tuple.from(1066L)), is(false)); assertThat(recordStore.recordExists(Tuple.from(1415L)), is(true)); + scrubAllIndexes(); commit(context); } } @Test - public void writeByteString() throws Exception { + void writeByteString() throws Exception { try (FDBRecordContext context = openContext()) { openBytesRecordStore(context); @@ -163,12 +224,13 @@ public void writeByteString() throws Exception { myrec1.mergeFrom(rec1.getRecord()); assertEquals(byteString(0, 1, 2), myrec1.getPkey()); assertEquals("foo", myrec1.getName()); + scrubAllIndexes(); commit(context); } } @Test - public void writeUuid() { + void writeUuid() { UUID uuid1 = UUID.fromString("710730ce-d9fd-417a-bb6e-27bcfefe3d4d"); UUID uuid2 = UUID.fromString("03b9221a-e61b-4bee-8c47-34e1248ed273"); @@ -194,7 +256,7 @@ public void writeUuid() { } @Test - public void writeNotUnionType() throws Exception { + void writeNotUnionType() throws Exception { try (FDBRecordContext context = openContext()) { openUnionRecordStore(context); @@ -208,9 +270,434 @@ public void writeNotUnionType() throws Exception { } } + @ParameterizedTest(name = "saveRecordsConcurrently[{0}]") + @BooleanSource + void saveRecordsConcurrently(boolean disableConcurrencyManagement) throws Exception { + final List> saved; + final FDBRecordStore.Builder storeBuilder; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context); + storeBuilder = recordStore.asBuilder().setDisableConcurrencyManagement(disableConcurrencyManagement); + recordStore = storeBuilder.open(); + + // Create 100 futures, each one saving a different record, and then run them concurrently. + // As they are each touching a different record, the operations should succeed regardless + // of whether the concurrency manager is disabled. + final List>> futures = IntStream.range(0, 100) + .mapToObj(id -> TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(id + 1000L) + .setNumValue3Indexed(id % 3) + .setNumValue2(id % 4) + .setStrValueIndexed((id % 2L == 0L) ? "even" : "odd") + .setNumValueUnique(id + 100) + .build() + ) + .map(recordStore::saveRecordAsync) + .toList(); + saved = AsyncUtil.getAll(futures).get(); + + commit(context); + } + try (FDBRecordContext context = openContext()) { + recordStore = storeBuilder.setContext(context).open(); + final List> loaded = AsyncUtil.getAll(saved.stream() + .map(FDBStoredRecord::getPrimaryKey) + .map(recordStore::loadRecordAsync) + .toList() + ).get(); + assertEquals(saved.size(), loaded.size(), "saved and loaded lists should have the same size"); + for (int i = 0; i < saved.size(); i++) { + FDBStoredRecord savedRecord = saved.get(i); + FDBStoredRecord loadedRecord = loaded.get(i); + assertEquals(savedRecord.getRecord(), loadedRecord.getRecord()); + } + assertEquals(saved.size(), recordStore.getSnapshotRecordCount().get()); + assertEquals(saved.size(), recordStore.getSnapshotRecordUpdateCount().get()); + + scrubAllIndexes(); + } + } + + @Test + void saveSameRecordConcurrently() throws Exception { + final FDBRecordStore.Builder storeBuilder; + final List> saved; + byte[] commitVersionstamp; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, metaDataBuilder -> metaDataBuilder.setStoreRecordVersions(true)); + storeBuilder = recordStore.asBuilder().setDisableConcurrencyManagement(false); + recordStore = storeBuilder.open(); + + // Create 100 futures, each one saving the same record (i.e., the same primary key), but with + // different values. Only one of these will succeed at the end, so + final List>> futures = IntStream.range(0, 100) + .mapToObj(id -> TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(1000L) + .setNumValue3Indexed(id % 3) + .setNumValue2(id) + .setStrValueIndexed((id % 2L == 0L) ? "even" : "odd") + .setNumValueUnique(id) + .build() + ) + .map(recordStore::saveRecordAsync) + .toList(); + saved = AsyncUtil.getAll(futures).get(); + + commit(context); + commitVersionstamp = Objects.requireNonNull(context.getVersionStamp()); + } + + try (FDBRecordContext context = openContext()) { + recordStore = storeBuilder.setContext(context).open(); + + final List> loaded = AsyncUtil.getAll(saved.stream() + .map(FDBStoredRecord::getPrimaryKey) + .map(recordStore::loadRecordAsync) + .toList() + ).get(); + assertEquals(saved.size(), loaded.size(), "saved and loaded lists should have the same size"); + boolean found = false; + for (int i = 0; i < saved.size(); i++) { + FDBStoredRecord savedRecord = saved.get(i); + FDBStoredRecord loadedRecord = loaded.get(i); + if (savedRecord.getRecord().equals(loadedRecord.getRecord())) { + found = true; + assertNotNull(loadedRecord.getVersion()); + assertNotNull(savedRecord.getVersion()); + assertEquals(savedRecord.getVersion().withCommittedVersion(commitVersionstamp), loadedRecord.getVersion()); + } + } + assertTrue(found, "no record found that matched original set"); + assertEquals(1L, recordStore.getSnapshotRecordCount().get()); + assertEquals(saved.size(), recordStore.getSnapshotRecordUpdateCount().get()); + + scrubAllIndexes(); + } + } + + @Test + void onlyOneConcurrentInsertSucceeds() throws Exception { + final FDBRecordStore.Builder storeBuilder; + final List> inserted; + byte[] commitVersionstamp; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, metaDataBuilder -> metaDataBuilder.setStoreRecordVersions(true)); + storeBuilder = recordStore.asBuilder().setDisableConcurrencyManagement(false); + recordStore = storeBuilder.open(); + + final List>> futures = IntStream.range(0, 100) + .mapToObj(id -> TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(1000L + (id % 10)) + .setNumValue3Indexed(id % 3) + .setNumValue2(id) + .setStrValueIndexed((id % 2L == 0L) ? "even" : "odd") + .build()) + .map(rec -> recordStore.insertRecordAsync(rec).handle((saved, err) -> { + if (err != null) { + if (err instanceof CompletionException) { + err = err.getCause(); + } + assertInstanceOf(RecordAlreadyExistsException.class, err); + return null; + } + return saved; + })) + .toList(); + final List> savedRecords = AsyncUtil.getAll(futures).get(); + inserted = savedRecords.stream().filter(Objects::nonNull).toList(); + assertEquals(10, inserted.size()); + final Set savedPrimaryKeys = inserted.stream().map(FDBStoredRecord::getPrimaryKey).collect(Collectors.toSet()); + assertEquals(10, savedPrimaryKeys.size()); + + assertEquals(10, recordStore.getSnapshotRecordCount().get()); + assertEquals(10, recordStore.getSnapshotRecordUpdateCount().get()); + + commit(context); + commitVersionstamp = Objects.requireNonNull(context.getVersionStamp()); + } + + try (FDBRecordContext context = openContext()) { + recordStore = storeBuilder.setContext(context).open(); + + inserted.forEach(insertedRecord -> { + final FDBStoredRecord stored = recordStore.loadRecord(insertedRecord.getPrimaryKey()); + assertNotNull(stored); + assertEquals(insertedRecord.getRecord(), stored.getRecord()); + assertEquals(Objects.requireNonNull(insertedRecord.getVersion()).withCommittedVersion(commitVersionstamp), stored.getVersion()); + }); + + scrubAllIndexes(); + } + } + + @Test + void deleteSameRecordConcurrently() throws Exception { + // Save a single record + final FDBRecordStore.Builder storeBuilder; + final FDBStoredRecord saved; + byte[] commitVersionstamp; + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, metaDataBuilder -> metaDataBuilder.setStoreRecordVersions(true)); + storeBuilder = recordStore.asBuilder().setDisableConcurrencyManagement(false); + recordStore = storeBuilder.open(); + + saved = recordStore.saveRecord(TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(1805L) + .setNumValue2(3) + .setStrValueIndexed("blah") + .setNumValue3Indexed(4) + .build()); + commit(context); + commitVersionstamp = Objects.requireNonNull(context.getVersionStamp()); + } + + // Attempt to delete that record from multiple places, interspersed with concurrent reads. + // Exactly one delete should succeed, and all the reads should occur either strictly before + // or strictly after the delete + try (FDBRecordContext context = openContext()) { + recordStore = storeBuilder.setContext(context).open(); + + // Fire off some reads before the first delete call + final List>> readFutures = new ArrayList<>(); + Stream.generate(() -> recordStore.loadRecordAsync(saved.getPrimaryKey())) + .limit(30) + .forEach(readFutures::add); + + // Now, issue multiple deletes + final List> deleteFutures = Stream.generate(() -> recordStore.deleteRecordAsync(saved.getPrimaryKey())) + .limit(30) + .toList(); + + // Add additional reads after the first delete + Stream.generate(() -> recordStore.loadRecordAsync(saved.getPrimaryKey())) + .limit(30) + .forEach(readFutures::add); + + final List> readResults = AsyncUtil.getAll(readFutures).get(); + final List deleteResults = AsyncUtil.getAll(deleteFutures).get(); + + // Exactly one of the deletes should return true + assertEquals(1, deleteResults.stream() + .filter(deleted -> deleted) + .count()); + + // All the read results should either be null (if they happened after the delete) or they should match the original record + readResults.stream() + .filter(Objects::nonNull) + .forEach(readRecord -> { + assertEquals(saved.getRecord(), readRecord.getRecord()); + assertEquals(Objects.requireNonNull(saved.getVersion()).withCommittedVersion(commitVersionstamp), readRecord.getVersion()); + }); + + assertEquals(0L, recordStore.getSnapshotRecordCount().get()); + + scrubAllIndexes(); + commit(context); + } + } + + private static class TaskState { + private int taskNumber; + private int updates; + private int completed; + @Nonnull + private final Map>> historyByRecord = new ConcurrentHashMap<>(); + } + + /** + * Create a random sequence of single-record operations and run them with concurrency. Each + * operation can do some kind of single-record operation (e.g., read or update one record). + * Those are spread across a small number of record primary keys. With each operation, we + * check to make sure (1) there are no exceptions hit executing the operation and (2) we + * the value is consistent with the operation history. Because of the concurrency, we + * are not able to fix the reads to a single possible value, but we should be able to assert + * that the value is one of a consistent set of values. + * + * @param seed a seed to use in the random number generator used to create test cases + * @throws Exception any problem hit while running the test + */ + @ParameterizedTest + @RandomSeedSource + void concurrentRecordOperationStressTest(long seed) throws Exception { + final int concurrentTasks = 100; + final int totalTasks = 1000; + final TaskState taskState = new TaskState(); + + final FDBRecordStore.Builder storeBuilder; + + try (FDBRecordContext context = openContext()) { + openSimpleRecordStore(context, metaDataBuilder -> { + metaDataBuilder.setStoreRecordVersions(true); + metaDataBuilder.setSplitLongRecords(true); + metaDataBuilder.removeIndex("MySimpleRecord$str_value_indexed"); + }); + // These tests fail if concurrence management is disabled. For this test, we also + // assert on the default behavior (that the store starts with concurrency management + // enabled). If this is changed, then this assert can be updated, but we still need + // to override the feature for this test + storeBuilder = recordStore.asBuilder(); + assertFalse(storeBuilder.isConcurrencyManagementDisabled()); + storeBuilder.setDisableConcurrencyManagement(false); + commit(context); + } + + try (FDBRecordContext context = openContext()) { + recordStore = storeBuilder.setContext(context).open(); + + // These tests will fail if we don't have the concurrency manager enabled. + // In theory, this could be an assumption, but making it an assert means that + // we are notified if we somehow lose coverage, and then we can decide if that's + // desirable or not + assertFalse(recordStore.asBuilder().isConcurrencyManagementDisabled()); + + final Random random = new Random(seed); + final Queue> tasks = new ArrayDeque<>(); + while (taskState.completed < totalTasks) { + while (tasks.size() < concurrentTasks && taskState.taskNumber < totalTasks) { + taskState.taskNumber++; + tasks.add(createRandomRecordOperation(random, taskState)); + } + // Wait for the head of the queue to complete, then remove the leading head of tasks that have already completed + tasks.peek().get(); + while (!tasks.isEmpty() && tasks.peek().isDone()) { + tasks.remove().get(); + taskState.completed++; + } + } + assertTrue(tasks.isEmpty()); + validateRecordsAfterRun(taskState); + commit(context); + } + + try (FDBRecordContext context = openContext()) { + recordStore = storeBuilder.setContext(context).open(); + validateRecordsAfterRun(taskState); + scrubAllIndexes(); + } + } + + private void validateRecordsAfterRun(@Nonnull TaskState taskState) throws Exception { + // The updates index should contain one udpate for every update started during the test + assertEquals(taskState.updates, recordStore.getSnapshotRecordUpdateCount().get()); + + // Make sure the most recent update is persisted for each record + int expectedCount = 0; + for (Map.Entry>> entry : taskState.historyByRecord.entrySet()) { + final Tuple primaryKey = entry.getKey(); + final Pair mostRecentUpdate = entry.getValue().peekFirst(); + @Nullable final Message expectedMessage = mostRecentUpdate == null ? null : mostRecentUpdate.getRight(); + FDBStoredRecord loaded = recordStore.loadRecord(primaryKey); + @Nullable final Message readMessage = loaded == null ? null : loaded.getRecord(); + assertEquals(expectedMessage, readMessage); + if (expectedMessage != null) { + expectedCount++; + } + } + assertEquals(expectedCount, recordStore.getSnapshotRecordCount().get()); + } + + @Nonnull + private CompletableFuture createRandomRecordOperation(@Nonnull Random random, @Nonnull TaskState taskState) { + final int taskNumber = taskState.taskNumber; + final int completed = taskState.completed; + double choice = random.nextDouble(); + final long recNo = random.nextLong(20); + final Tuple primaryKey = Tuple.from(recNo); + Deque> recordHistory = taskState.historyByRecord.computeIfAbsent(primaryKey, ignored -> new ConcurrentLinkedDeque<>()); + if (choice < 0.3) { + // Read the record at this primary key. + return recordStore.loadRecordAsync(primaryKey).thenApply(stored -> { + Set possibleValues = possibleValuesForRecord(recordHistory, completed); + Message storedRec = stored == null ? null : stored.getRecord(); + assertThat(storedRec, in(possibleValues)); + if (stored != null) { + assertEquals(primaryKey, stored.getPrimaryKey()); + } + return null; + }); + } else if (choice < 0.5) { + // Check if the record at this primary key exists + return recordStore.recordExistsAsync(primaryKey).thenApply(exists -> { + Set possibleValues = possibleValuesForRecord(recordHistory, completed); + if (exists) { + assertTrue(anyNonNull(possibleValues), "record exists, so there should be at least one non-null possible value"); + } else { + assertTrue(canBeNull(possibleValues), "record does not exist, so null must be a possible value"); + } + return null; + }); + } else if (choice < 0.6) { + // Preload the record from the database. This will populate the preload cache with a value + // read from the database. Later reads from the database will use this preloaded value if + // it is set, so this is important for making sure that we later clean up that state. That is, + // the presence of "preload" events in the history should not change the behavior of the other + // operations + return recordStore.preloadRecordAsync(primaryKey); + } else if (choice < 0.9) { + // Insert a new value for a record + TestRecords1Proto.MySimpleRecord rec = TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(recNo) + .setNumValueUnique(taskNumber) + .setStrValueIndexed(random.nextDouble() < 0.05 ? longString : "blah") + .setNumValue3Indexed(random.nextInt(3)) + .build(); + recordHistory.addFirst(Pair.of(taskNumber, rec)); + taskState.updates++; + return recordStore.saveRecordAsync(rec).thenApply(ignored -> null); + } else { + // Delete the record. + recordHistory.addFirst(Pair.of(taskNumber, null)); + return recordStore.deleteRecordAsync(primaryKey).thenApply(deleted -> { + final Set possibleValues = possibleValuesForRecord(recordHistory, completed); + if (deleted) { + assertTrue(anyNonNull(possibleValues), "record previously existed, so some non-null value must be possible"); + } + return null; + }); + } + } + + /** + * Compute the set of possible values for a record given its history. The history array should be sorted + * in reverse chronological order with the latest writes to the history coming at the front. When a + * task is started, some tail of the history is already completed. Any read must therefore be some + * value that was either (1) begun after all the completed tasks or (2) was the final write completed. + * Note that we're relying here on the way that write locks are managed in the {@link com.apple.foundationdb.record.locking.LockRegistry}. + * That is to say, that we always enqueue later writes on top of older writes, so they end up executing + * in the order in which they are created. + * + * @param history the history of values for the record in descending version order + * @param completedAtStart the final update that is guaranteed to have completed after the read started + * @return a set of possible values the read could be + */ + @Nonnull + private static Set possibleValuesForRecord(@Nonnull Deque> history, int completedAtStart) { + Set possibleValues = new HashSet<>(); + boolean foundOldest = false; + for (final Pair pair : history) { + possibleValues.add(pair.getRight()); + if (Objects.requireNonNull(pair.getLeft()) < completedAtStart) { + foundOldest = true; + break; + } + } + if (!foundOldest) { + possibleValues.add(null); + } + return possibleValues; + } + + private static boolean canBeNull(@Nonnull Set possibleValues) { + return possibleValues.contains(null); + } + + private static boolean anyNonNull(@Nonnull Set possibleValues) { + return possibleValues.stream().anyMatch(Objects::nonNull); + } + @Test - public void readPreloaded() throws Exception { - byte[] versionstamp; + void readPreloaded() throws Exception { + byte[] commitVersionstamp; try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -220,8 +707,8 @@ public void readPreloaded() throws Exception { recordStore.saveRecord(rec); commit(context); - versionstamp = context.getVersionStamp(); - assertNotNull(versionstamp); + commitVersionstamp = context.getVersionStamp(); + assertNotNull(commitVersionstamp); } try (FDBRecordContext context = openContext()) { @@ -232,7 +719,7 @@ public void readPreloaded() throws Exception { assertNotNull(record); assertSame(TestRecords1Proto.MySimpleRecord.getDescriptor(), record.getRecordType().getDescriptor()); assertEquals(1066L, record.getRecord().getField(TestRecords1Proto.MySimpleRecord.getDescriptor().findFieldByNumber(TestRecords1Proto.MySimpleRecord.REC_NO_FIELD_NUMBER))); - assertEquals(FDBRecordVersion.complete(versionstamp, 0), record.getVersion()); + assertEquals(FDBRecordVersion.complete(commitVersionstamp, 0), record.getVersion()); FDBExceptions.FDBStoreException e = assertThrows(FDBExceptions.FDBStoreException.class, context::commit); assertNotNull(e.getCause()); @@ -243,7 +730,7 @@ public void readPreloaded() throws Exception { } @Test - public void readMissingPreloaded() throws Exception { + void readMissingPreloaded() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); // 4488 does not exist @@ -262,7 +749,7 @@ public void readMissingPreloaded() throws Exception { } @Test - public void readYourWritesPreloaded() throws Exception { + void readYourWritesPreloaded() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -288,7 +775,7 @@ public void readYourWritesPreloaded() throws Exception { } @Test - public void deletePreloaded() throws Exception { + void deletePreloaded() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -309,7 +796,7 @@ public void deletePreloaded() throws Exception { } @Test - public void deleteAllPreloaded() throws Exception { + void deleteAllPreloaded() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -330,7 +817,7 @@ public void deleteAllPreloaded() throws Exception { } @Test - public void saveOverPreloaded() throws Exception { + void saveOverPreloaded() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -362,7 +849,7 @@ public void saveOverPreloaded() throws Exception { } @Test - public void preloadNonExisting() throws Exception { + void preloadNonExisting() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); @@ -372,7 +859,7 @@ public void preloadNonExisting() throws Exception { } @Test - public void delete() throws Exception { + void delete() throws Exception { try (FDBRecordContext context = openContext()) { openSimpleRecordStore(context); diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreTest.java index b9dc0af577..004f9f7667 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStoreTest.java @@ -55,6 +55,8 @@ import com.apple.foundationdb.record.metadata.expressions.KeyExpression; import com.apple.foundationdb.record.provider.common.RecordSerializationException; import com.apple.foundationdb.record.provider.common.RecordSerializer; +import com.apple.foundationdb.record.provider.foundationdb.concurrency.FDBRecordStoreConcurrencyManager; +import com.apple.foundationdb.record.provider.foundationdb.concurrency.NoOpConcurrencyManager; import com.apple.foundationdb.record.provider.foundationdb.keyspace.KeySpacePath; import com.apple.foundationdb.record.query.RecordQuery; import com.apple.foundationdb.record.query.expressions.Comparisons; @@ -1532,7 +1534,8 @@ public FDBRecordStore build() { getIndexMaintainerRegistry(), getIndexMaintenanceFilter(), getPipelineSizer(), getStoreStateCache(), getStateCacheabilityOnOpen(), getUserVersionChecker(), - getBypassFullStoreLockReason(), getPlanSerializationRegistry()) { + getBypassFullStoreLockReason(), getPlanSerializationRegistry(), + isConcurrencyManagementDisabled() ? NoOpConcurrencyManager.instance() : new FDBRecordStoreConcurrencyManager(getSubspaceProvider(), getContext())) { @Nonnull @Override protected CompletableFuture getRecordCountForRebuildIndexes( diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStoreTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStoreTest.java index 75b9326e99..2fc59e5d5a 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStoreTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/FDBTypedRecordStoreTest.java @@ -20,6 +20,7 @@ package com.apple.foundationdb.record.provider.foundationdb; +import com.apple.foundationdb.async.AsyncUtil; import com.apple.foundationdb.record.RecordCursorIterator; import com.apple.foundationdb.record.RecordMetaDataProto; import com.apple.foundationdb.record.StoreIsFullyLockedException; @@ -35,16 +36,21 @@ import com.apple.foundationdb.record.test.TestKeySpace; import com.apple.foundationdb.record.test.TestKeySpacePathManagerExtension; import com.apple.foundationdb.tuple.Tuple; +import com.apple.test.BooleanSource; import com.apple.test.Tags; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.junit.jupiter.params.ParameterizedTest; import javax.annotation.Nonnull; import java.util.ArrayList; import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; +import java.util.stream.IntStream; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -320,4 +326,83 @@ void testClearFullStoreLock() { } } + @ParameterizedTest(name = "disableConcurrencyManager[{0}]") + @BooleanSource + void disableConcurrencyManager(boolean disableConcurrencyManager) throws Exception { + FDBTypedRecordStore.Builder builder = BUILDER.copyBuilder() + .setKeySpacePath(path) + .setDisableConcurrencyManagement(disableConcurrencyManager); + assertEquals(disableConcurrencyManager, builder.isConcurrencyManagementDisabled()); + + // Validate that creating a store, creating it, and then turning it back into a builder + // preserves the concurrency management setting + try (FDBRecordContext context = fdb.openContext()) { + recordStore = builder.setContext(context).create(); + context.commit(); + builder = recordStore.asBuilder(); + } + assertEquals(disableConcurrencyManager, builder.isConcurrencyManagementDisabled()); + + // Basic smoke test of concurrency manager validation. The typed class delegates to the + // + try (FDBRecordContext context = fdb.openContext()) { + recordStore = builder.setContext(context).open(); + + final List>> futures = IntStream.range(0, 100) + .mapToObj(id -> recordStore.saveRecordAsync( + TestRecords1Proto.MySimpleRecord.newBuilder() + .setRecNo(100) + .setNumValueUnique(id) + .setNumValue3Indexed(id % 3) + .setStrValueIndexed((id % 2 == 0) ? "even" : "odd") + .build() + )) + .toList(); + final CompletableFuture>> futureList = AsyncUtil.getAll(futures); + + if (disableConcurrencyManager) { + try { + futureList.get(); + } catch (ExecutionException e) { + // With no concurrency manager, this can throw an exception. Not a lot we can assert here, so just ignore + } + } else { + // Should not throw an exception + futureList.get(); + } + + // Basic validation. Read the one record and make sure it is aligned with one of the ones that was written + FDBStoredRecord loaded = recordStore.loadRecord(Tuple.from(100L)); + assertNotNull(loaded); + TestRecords1Proto.MySimpleRecord loadedRec = loaded.getRecord(); + assertThat(loadedRec.getRecNo()) + .isEqualTo(100L); + assertThat(loadedRec.getNumValueUnique()) + .isGreaterThanOrEqualTo(0) + .isLessThan(100); + assertThat(loadedRec.getNumValue3Indexed()) + .isEqualTo(loadedRec.getNumValueUnique() % 3); + assertThat(loadedRec.getStrValueIndexed()) + .isEqualTo((loadedRec.getNumValueUnique() % 2 == 0) ? "even" : "odd"); + context.commit(); + } + + // Check the num_value_unique index + try (OnlineIndexScrubber scrubber = OnlineIndexScrubber.newBuilder() + .setRecordStore(recordStore.getUntypedRecordStore()) + .setIndex(recordStore.getRecordMetaData().getIndex("MySimpleRecord$num_value_unique")) + .setScrubbingPolicy(OnlineIndexScrubber.ScrubbingPolicy.newBuilder().setAllowRepair(true).build()) + .build()) { + long dangling = scrubber.scrubDanglingIndexEntries(); + long missing = scrubber.scrubMissingIndexEntries(); + + if (!disableConcurrencyManager) { + // With no concurrency manager, it's not guaranteed that the index entries line up + assertThat(dangling) + .isZero(); + assertThat(missing) + .isZero(); + } + } + } } diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/MultidimensionalIndexTestBase.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/MultidimensionalIndexTestBase.java index 8a73edcb3b..5746b6c96a 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/MultidimensionalIndexTestBase.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/MultidimensionalIndexTestBase.java @@ -82,6 +82,8 @@ import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.protobuf.Descriptors; import com.google.protobuf.Message; +import org.hamcrest.MatcherAssert; +import org.hamcrest.Matchers; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Tag; import org.opentest4j.AssertionFailedError; @@ -290,9 +292,14 @@ static Function getRecordWithNullGenerator(@Nonnull Random ran } public void loadRecords(final boolean useAsync, final boolean withNulls, @Nonnull final RecordMetaDataHook hook, - final long seed, final List calendarNames, final int numSamples) { + final long seed, final List calendarNames, final int numSamples) { final Random random = new Random(seed); final var recordGenerator = withNulls ? getRecordWithNullGenerator(random, calendarNames) : getRecordGenerator(random, calendarNames); + loadRecords(useAsync, hook, numSamples, recordGenerator); + } + + public void loadRecords(final boolean useAsync, @Nonnull final RecordMetaDataHook hook, final int numSamples, + @Nonnull final Function recordGenerator) { if (useAsync) { Assertions.assertDoesNotThrow(() -> batch(hook, numSamples, 500, recNo -> recordStore.saveRecord(recordGenerator.apply(recNo)))); } else { @@ -491,6 +498,26 @@ void basicReadWithNullsTest(final boolean useAsync, @Nonnull final String storag } } + void basicReadWithOverwritesTest(final boolean useAsync, @Nonnull final String storage, final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { + final RecordMetaDataHook additionalIndex = metaDataBuilder -> addMultidimensionalIndex(metaDataBuilder, storage, + storeHilbertValues, useNodeSlotIndex); + @Nonnull Random r = new Random(0); + @Nonnull Function baseGenerator = getRecordGenerator(r, ImmutableList.of("business", "home")); + loadRecords(useAsync, additionalIndex, 500, recNo -> baseGenerator.apply(recNo % 5)); + try (FDBRecordContext context = openContext()) { + openRecordStore(context, additionalIndex); + for (long l = 0; l < 5; l++) { + FDBStoredRecord rec = recordStore.loadRecord(Tuple.from(null, l)); + assertNotNull(rec); + TestRecordsMultidimensionalProto.MyMultidimensionalRecord.Builder recordBuilder = + TestRecordsMultidimensionalProto.MyMultidimensionalRecord.newBuilder(); + recordBuilder.mergeFrom(rec.getRecord()); + MatcherAssert.assertThat(recordBuilder.getCalendarName(), Matchers.either(Matchers.equalTo("business")).or(Matchers.equalTo("home"))); + } + commit(context); + } + } + void indexReadTest(final boolean useAsync, final long seed, final int numRecords, @Nonnull final String storage, final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { final RecordMetaDataHook additionalIndexes = @@ -549,7 +576,7 @@ void indexReadTest(final boolean useAsync, final long seed, final int numRecords } void indexReadWithNullsTest(final boolean useAsync, final long seed, final int numRecords, @Nonnull final String storage, - final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { + final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { RecordMetaDataHook additionalIndexes = metaDataBuilder -> { addCalendarNameStartEpochIndex(metaDataBuilder); @@ -606,6 +633,68 @@ void indexReadWithNullsTest(final boolean useAsync, final long seed, final int n Assertions.assertEquals(expectedResults, actualResults); } + void indexReadWithOverwritesTest(final boolean useAsync, final long seed, final int numRecords, @Nonnull final String storage, + final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { + RecordMetaDataHook additionalIndexes = + metaDataBuilder -> { + addCalendarNameStartEpochIndex(metaDataBuilder); + addMultidimensionalIndex(metaDataBuilder, storage, storeHilbertValues, useNodeSlotIndex); + }; + final Random random = new Random(seed); + final Function baseRecordGenerator = getRecordGenerator(random, ImmutableList.of("business")); + loadRecords(useAsync, additionalIndexes, numRecords, recNo -> baseRecordGenerator.apply(recNo % 10)); + + final long intervalStartInclusive = epochMean + 3600L; + final long intervalEndInclusive = epochMean + 5L * 3600L; + final RecordQueryIndexPlan indexPlan = + new RecordQueryIndexPlan("EventIntervals", + new HypercubeScanParameters("business", + (Long)null, intervalEndInclusive, + intervalStartInclusive, null), + false); + Set actualResults = getResults(additionalIndexes, indexPlan); + MatcherAssert.assertThat(actualResults.size(), Matchers.lessThanOrEqualTo(10)); + + final QueryComponent filter = + Query.and( + Query.field("calendar_name").equalsValue("business"), + Query.or( + Query.field("start_epoch").isNull(), + Query.field("start_epoch").lessThanOrEquals(intervalEndInclusive)), + Query.field("end_epoch").greaterThanOrEquals(intervalStartInclusive)); + + RecordQuery query = RecordQuery.newBuilder() + .setRecordType("MyMultidimensionalRecord") + .setFilter(filter) + .setIndexQueryabilityFilter(noMultidimensionalIndexes) + .build(); + final RecordQueryPlan plan = planQuery(query); + final Set expectedResults = getResults(additionalIndexes, plan); + Assertions.assertEquals(expectedResults, actualResults); + + // run an un-hinted query -- make sure the planner picks up the md-index + query = RecordQuery.newBuilder() + .setRecordType("MyMultidimensionalRecord") + .setFilter(filter) + .build(); + final RecordQueryPlan mdPlan = planQuery(query); + + assertMatchesExactly(mdPlan, + unorderedPrimaryKeyDistinctPlan( + unorderedUnionPlan( + indexPlan() + .where(indexName("EventIntervals")) + .and(indexScanParameters( + multidimensional() + .where(prefix(range("[[business],[business]]"))) + .and(dimensions(range("([null],[1690378647]]"), range("[[1690364247],>"))) + .and(suffix(unbounded())))), + indexPlan().where(indexName("calendarNameStartEpoch")) + ))); + actualResults = getResults(additionalIndexes, mdPlan); + Assertions.assertEquals(expectedResults, actualResults); + } + void indexReadWithNullsAndMinsTest1(final boolean useAsync) throws Exception { RecordMetaDataHook additionalIndexes = metaDataBuilder -> { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SimpleMultidimensionalIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SimpleMultidimensionalIndexTest.java index f203aa1d04..c65e6ccae7 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SimpleMultidimensionalIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SimpleMultidimensionalIndexTest.java @@ -174,6 +174,13 @@ void basicReadWithNullsTest(@Nonnull final String storage, final boolean storeHi super.basicReadWithNullsTest(false, storage, storeHilbertValues, useNodeSlotIndex); } + @ParameterizedTest + @MethodSource("argumentsForBasicReads") + void basicReadWithOverwritesTest(@Nonnull final String storage, final boolean storeHilbertValues, + final boolean useNodeSlotIndex) throws Exception { + super.basicReadWithOverwritesTest(false, storage, storeHilbertValues, useNodeSlotIndex); + } + @ParameterizedTest @MethodSource("argumentsForBasicReads") void deleteWhereTest(@Nonnull final String storage, final boolean storeHilbertValues, final boolean useNodeSlotIndex) @@ -202,6 +209,13 @@ void indexReadWithNullsTest(final long seed, final int numRecords, @Nonnull fina super.indexReadWithNullsTest(false, seed, numRecords, storage, storeHilbertValues, useNodeSlotIndex); } + @ParameterizedTest + @MethodSource("argumentsForIndexReads") + void indexReadWithOverwritesTest(final long seed, final int numRecords, @Nonnull final String storage, + final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { + super.indexReadWithOverwritesTest(false, seed, numRecords, storage, storeHilbertValues, useNodeSlotIndex); + } + @ParameterizedTest @MethodSource("argumentsForIndexReads") void indexReadIsNullTest(final long seed, final int numRecords, @Nonnull final String storage, 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 25f8d0a3fc..90a706084b 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 @@ -57,6 +57,7 @@ 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.SplitHelper; 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; @@ -66,29 +67,43 @@ import com.apple.foundationdb.record.query.expressions.Query; import com.apple.foundationdb.record.slidingwindowvector.TestRecordsSlidingWindowVectorProto.SlidingWindowVectorRecord; import com.apple.foundationdb.tuple.Tuple; +import com.apple.test.BooleanSource; +import com.apple.test.RandomSeedSource; import com.apple.test.Tags; import com.google.common.collect.ImmutableList; +import com.google.errorprone.annotations.CanIgnoreReturnValue; import com.google.protobuf.Any; import com.google.protobuf.ByteString; import com.google.protobuf.Message; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; import javax.annotation.Nonnull; import javax.annotation.Nullable; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; import java.util.List; +import java.util.Random; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import java.util.stream.LongStream; import static com.apple.foundationdb.record.provider.foundationdb.indexes.SlidingWindowTestHelpers.SlidingWindowAssert.assertThat; import static com.apple.foundationdb.record.provider.foundationdb.indexes.SlidingWindowTestHelpers.makeVector; import static com.apple.foundationdb.record.provider.foundationdb.indexes.SlidingWindowTestHelpers.sampleVector; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -141,14 +156,24 @@ private void rec(long recNo, long relevance) { } private void rec(long recNo, String zone, String category, long relevance, long score, HalfRealVector vector) { - recordStore.saveRecord(SlidingWindowVectorRecord.newBuilder() + recordStore.saveRecord(createRecord(recNo, zone, category, relevance, score, vector)); + } + + @Nonnull + private SlidingWindowVectorRecord createRecord(long recNo, long relevance) { + return createRecord(recNo, "z", "c", relevance, 0, sampleVector()); + } + + @Nonnull + private SlidingWindowVectorRecord createRecord(long recNo, String zone, String category, long relevance, long score, HalfRealVector vector) { + return SlidingWindowVectorRecord.newBuilder() .setRecNo(recNo) .setZone(zone) .setCategory(category) .setRelevance(relevance) .setScore(score) .setVectorData(ByteString.copyFrom(vector.getRawData())) - .build()); + .build(); } private void deleteRec(long recNo) { @@ -438,6 +463,190 @@ void deleteAllAndRefill() throws Exception { } } + // ===== Concurrent manipulation tests ===== + + /** + * Validate concurrent saves work if each save is for a unique record. As each save is unique, + * the only place where intra-transaction concurrency can cause a problem is within the + * index maintainer itself. The {@link SlidingWindowIndexMaintainer} grabs a write lock over + * the window subspace for that reason, which means we effectively serialize access to that + * area. For that reason, we expect consistent results whether the store's concurrency + * manager is enabled or not. + * + * @param disableConcurrencyManagement whether to disable the store's concurrency manager in the test + * @throws Exception any exception thrown during the test + */ + @ParameterizedTest + @BooleanSource + void concurrentInsertsToDistinctKeys(boolean disableConcurrencyManagement) throws Exception { + final int totalRecords = 150; + final int concurrency = 50; + final int windowSize = 10; + try (FDBRecordContext context = openContext()) { + openStore(context, windowSize, Direction.DESC); + recordStore = recordStore.asBuilder().setDisableConcurrencyManagement(disableConcurrencyManagement).open(); + saveAllWithConcurrency(totalRecords, concurrency, + id -> createRecord(id, 300L + (id % 2 == 0 ? 1L : -1L) * id)); + + assertThat(slidingWindow()) + .hasSizeOf(windowSize) + .underlyingHnsw() + // Should contain the limit (10) largest even records, as those the most relevant + .containsInAnyOrder(LongStream.range(totalRecords - 2 * windowSize, totalRecords).filter(l -> l % 2 == 0).toArray()); + } + } + + /** + * Validate what happens if we have concurrent saves and some of those are overwrites of existing records. + * This can sometimes hit a deadlock that we have more explicit testing for in + * {@link #concurrentlySaveOtherRecordWhileUpdatingBoundaryKey(boolean)}. + * + * @param disableConcurrencyManagement whether to disable the store's concurrency manager in the test + * @throws Exception any exception thrown during the test + */ + @ParameterizedTest(name = "concurrentInsertsToKeysWithOverwrites[disableConcurrencyManagement={0}]") + @BooleanSource + void concurrentInsertsToKeysWithOverwrites(boolean disableConcurrencyManagement) throws Exception { + final int totalRecords = 300; + final int concurrency = 75; + final int windowSize = 5; + try (FDBRecordContext context = openContext()) { + openStore(context, windowSize, Direction.DESC); + recordStore = recordStore.asBuilder().setDisableConcurrencyManagement(disableConcurrencyManagement).open(); + + try { + saveAllWithConcurrency(totalRecords, concurrency, + id -> createRecord(id % 20, 300 + (long)(id % 2 == 0 ? 1 : -1) * id)); + } catch (TimeoutException e) { + // Should only get a timeout if we disable concurrency management + assertFalse(disableConcurrencyManagement); + Assumptions.assumeFalse(true); + } catch (ExecutionException e) { + // Should only get a FoundSplitWithoutStartException if concurrency management is disabled + assertInstanceOf(SplitHelper.FoundSplitWithoutStartException.class, e.getCause()); + assertTrue(disableConcurrencyManagement); + Assumptions.assumeFalse(true); + } + + assertThat(slidingWindow()) + .hasSizeOf(5) + .underlyingHnsw() + // There are 20 unique keys. The top 5 even ones are the ones relevant enough to be indexed with the window as is + .containsInAnyOrder(LongStream.range(10, 20).filter(l -> l % 2 == 0).toArray()); + } + } + + /** + * Validate that we can update records concurrently as long as we don't have a deadlock on reading + * boundary key. This test achieves this by inserting 40 records, and then only updating the 5 most + * and least relevant. This means the boundary key always points to a record that is in the middle and + * is therefore not updated. This means we avoid the concurrency problem alluded to in + * {@link #concurrentlySaveOtherRecordWhileUpdatingBoundaryKey(boolean)}. This test makes sure we have + * sensible outcomes in such a regime. + * + * @param seed the seed to use when generating random values + * @throws Exception any exception thrown during the test + */ + @ParameterizedTest + @RandomSeedSource + void concurrentInsertsToKeysWithOverwritesConstantBoundary(long seed) throws Exception { + final Random random = new Random(seed); + final int totalRecords = 300; + final int concurrency = 75; + final int windowSize = 20; + try (FDBRecordContext context = openContext()) { + openStore(context, windowSize, Direction.DESC); + + // Save 20 records with relevance < 100 and 20 with relevance >= 100 + for (int i = 0; i < 40; i++) { + rec(i, i * 5); + } + // With window size 5, the boundary key is relevance 100. + saveAllWithConcurrency(totalRecords, concurrency, id -> { + // Mutate either the 5 most or 5 least relevant records + int recNo = random.nextInt(10); + int relevance = random.nextInt(50); + if (recNo >= 5) { + recNo = 39 - recNo; + relevance += 150; + } + return createRecord(recNo, relevance); + }); + + assertThat(slidingWindow()) + .hasSizeOf(20) + .underlyingHnsw() + // The top 20 most relevant records should still be the top 20 by id + .containsInAnyOrder(LongStream.range(20, 40).toArray()); + } + } + + @CanIgnoreReturnValue + @Nonnull + private List> saveAllWithConcurrency(final int totalRecords, final int concurrency, @Nonnull Function recordGenerator) throws Exception { + final Deque>> futures = new ArrayDeque<>(); + final List> records = new ArrayList<>(); + int startedSaves = 0; + while (records.size() < totalRecords) { + while (startedSaves < totalRecords && futures.size() < concurrency) { + futures.addLast(recordStore.saveRecordAsync(recordGenerator.apply(startedSaves))); + startedSaves++; + } + futures.peekFirst().get(1, TimeUnit.SECONDS); + while (!futures.isEmpty() && futures.peekFirst().isDone()) { + records.add(futures.pollFirst().get(1, TimeUnit.SECONDS)); + } + } + return records; + } + + /** + * Check what happens if we have two concurrent updates where one of them is the boundary key record. + * + * @param disableConcurrencyManagement whether the store's concurrency management should be disabled + * @throws Exception any exception encountered by the test + */ + @ParameterizedTest(name = "concurrentlySaveOtherRecordWhileUpdatingBoundaryKey[disableConcurrencyManagement={0}]") + @BooleanSource + void concurrentlySaveOtherRecordWhileUpdatingBoundaryKey(boolean disableConcurrencyManagement) throws Exception { + try (FDBRecordContext context = openContext()) { + openStore(context, 2, Direction.DESC); + recordStore = recordStore.asBuilder().setDisableConcurrencyManagement(disableConcurrencyManagement).open(); + + rec(1, 100); + rec(2, 200); + rec(3, 300); + + // Create a new record (4) and also update the pre-existing record (2) that is on the boundary. They both + // end up adjusting the window, one of them to 210 and the other to 190. + // + // If we have not disabled store concurrency management, then the first save grab a write lock on record 4, + // then when updating the index, it grabs a write lock on the window space and a read lock on record 2. The second save needs to + // grab a write lock on record 2, after which it will grab a write lock on the window space. So there's a likely + // thread ordering of: + // + // 1. f1 grabs write locks on record 4 and the index window space + // 2. f2 brags a write lock on record 2 and then waits for f1 to release the index window space lock + // 3. f1 attempts to read the boundary key, so it waits for f2 to to release the lock on record 2 + // + // At this point, they're stuck. + final CompletableFuture> f1 = recordStore.saveRecordAsync(createRecord(4, 210)); + final CompletableFuture> f2 = recordStore.saveRecordAsync(createRecord(2, 190)); + try { + CompletableFuture.allOf(f1, f2).get(1, TimeUnit.SECONDS); + } catch (TimeoutException e) { + // Should only time out here if concurrency management is enabled + assertFalse(disableConcurrencyManagement); + Assumptions.assumeFalse(true); + } + + assertThat(slidingWindow()) + .hasSizeOf(2) + .underlyingHnsw() + .containsInAnyOrder(3L, 4L); + } + } + // ===== Re-election tests ===== @Test diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowTestHelpers.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowTestHelpers.java index f0efb6ac2f..33768b94a9 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowTestHelpers.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlidingWindowTestHelpers.java @@ -156,7 +156,7 @@ public static Set scanIndexRecNos(@Nonnull final FDBRecordStore recordStor final HalfRealVector queryVector = makeVector(0.5f, 0.5f, 0.5f, 0.5f); final double actualDistance = new Metric.EuclideanMetric().distance(queryVector.getData(), sampleVector().getData()); - final int limit = (int)(3 /*safety*/ + actualDistance); // overestimate limit to guarantee retrieval of all vectors. + final int limit = (int)(100 + actualDistance); // overestimate limit to guarantee retrieval of all vectors. final TupleRange range = groupingKey == null ? TupleRange.ALL : TupleRange.allOf(groupingKey); final VectorIndexScanBounds bounds = new VectorIndexScanBounds( diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlowMultidimensionalIndexTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlowMultidimensionalIndexTest.java index d8eae11a0b..9cd7e0e435 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlowMultidimensionalIndexTest.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/SlowMultidimensionalIndexTest.java @@ -166,6 +166,12 @@ void basicReadWithNulls(@Nonnull final String storage, final boolean storeHilber super.basicReadWithNullsTest(true, storage, storeHilbertValues, useNodeSlotIndex); } + @ParameterizedTest + @MethodSource("argumentsForBasicReads") + void basicReadWithOverwrites(@Nonnull final String storage, final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { + super.basicReadWithOverwritesTest(true, storage, storeHilbertValues, useNodeSlotIndex); + } + @ParameterizedTest @MethodSource("argumentsForBasicReads") void deleteWhereTest(@Nonnull final String storage, final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { @@ -193,6 +199,13 @@ void indexReadWithNullsTest(final long seed, final int numRecords, @Nonnull fina super.indexReadWithNullsTest(true, seed, numRecords, storage, storeHilbertValues, useNodeSlotIndex); } + @ParameterizedTest + @MethodSource("argumentsForIndexReads") + void indexReadWithOverwritesTest(final long seed, final int numRecords, @Nonnull final String storage, + final boolean storeHilbertValues, final boolean useNodeSlotIndex) throws Exception { + super.indexReadWithOverwritesTest(true, seed, numRecords, storage, storeHilbertValues, useNodeSlotIndex); + } + @ParameterizedTest @MethodSource("argumentsForIndexReads") void indexReadIsNullTest(final long seed, final int numRecords, @Nonnull final String storage, diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexEngineTestSuite.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexEngineTestSuite.java index 3e608b56c7..5f41f7481b 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexEngineTestSuite.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexEngineTestSuite.java @@ -71,11 +71,13 @@ import javax.annotation.Nonnull; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Random; import java.util.Set; +import java.util.function.Function; import java.util.stream.IntStream; import java.util.stream.Stream; @@ -130,6 +132,29 @@ void basicWriteReadTest(final long seed, final boolean useAsync) throws Exceptio } } + @ParameterizedTest + @MethodSource("randomSeedsWithAsync") + void basicWriteReadWithOverwritesTest(final long seed, final boolean useAsync) throws Exception { + final Random random = new Random(seed); + final Function baseRecordGenerator = getRecordGenerator(random, 0.3); + final List> savedRecords = + saveRandomRecords(useAsync, this::addVectorIndexes, 1000, recNo -> baseRecordGenerator.apply(recNo % 10)); + final Map> candidatesByPrimaryKey = new HashMap<>(); + savedRecords.forEach(savedRecord -> + candidatesByPrimaryKey.computeIfAbsent(savedRecord.getPrimaryKey(), k -> Sets.newHashSet()).add(savedRecord.getRecord())); + try (final FDBRecordContext context = openContext()) { + openRecordStore(context, this::addVectorIndexes); + for (final Map.Entry> entry : candidatesByPrimaryKey.entrySet()) { + final FDBStoredRecord loadedRecord = + recordStore.loadRecord(entry.getKey()); + + assertThat(loadedRecord).isNotNull(); + assertThat(loadedRecord.getRecord()).isIn(entry.getValue()); + } + commit(context); + } + } + @ParameterizedTest @MethodSource("randomSeedsWithAsyncAndLimit") void basicWriteIndexReadWithContinuationTest(final long seed, final boolean useAsync, final int limit) throws Exception { @@ -153,6 +178,43 @@ void basicWriteIndexReadWithContinuationTest(final long seed, final boolean useA checkResults(indexPlan, limit, expectedResults); } + @ParameterizedTest + @MethodSource("randomSeedsWithAsyncAndLimit") + void basicWriteIndexReadWithOverwritesTest(final long seed, final boolean useAsync, final int limit) throws Exception { + final int k = 10; + final Random random = new Random(seed); + final HalfRealVector queryVector = randomHalfVector(random, 128); + + // Save 1000 random records, but with only 50 unique primary keys + final Function baseRecordGenerator = getRecordGenerator(random, 0.0); + final List> savedRecords = + saveRandomRecords(useAsync, this::addUngroupedVectorIndex, 1000, recNo -> baseRecordGenerator.apply(random.nextLong(50))); + final Map> candidatesByPrimaryKey = new HashMap<>(); + savedRecords.forEach(saved -> candidatesByPrimaryKey.computeIfAbsent(saved.getPrimaryKey(), ignore -> Sets.newHashSet()).add(saved.getRecord())); + + final List> loadedRecords; + try (FDBRecordContext context = openContext()) { + openRecordStore(context, this::addUngroupedVectorIndex); + loadedRecords = recordStore.scanRecords(null, ScanProperties.FORWARD_SCAN).asList().get(); + assertThat(loadedRecords) + .hasSizeLessThanOrEqualTo(50) + .allSatisfy(loadedRecord -> + assertThat(loadedRecord.getRecord()).isIn(candidatesByPrimaryKey.get(loadedRecord.getPrimaryKey()))); + } + + final Set expectedResults = + sortByDistances(loadedRecords, queryVector, Metric.EUCLIDEAN_METRIC).stream() + .limit(k) + .map(nodeReferenceWithDistance -> + nodeReferenceWithDistance.getPrimaryKey().getLong(0)) + .collect(ImmutableSet.toImmutableSet()); + + final RecordQueryIndexPlan indexPlan = + createIndexPlan(queryVector, k, "UngroupedVectorIndex"); + + checkResults(indexPlan, limit, expectedResults); + } + private void checkResults(@Nonnull final RecordQueryIndexPlan indexPlan, final int limit, @Nonnull final Set expectedResults) throws Exception { diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexTestBase.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexTestBase.java index 7d07a63e34..ac62355052 100644 --- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexTestBase.java +++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/indexes/VectorIndexTestBase.java @@ -307,6 +307,13 @@ protected List> saveRandomRecords(final boolean useAsyn final int numRecords, final double nullProbability) throws Exception { final var recordGenerator = getRecordGenerator(random, nullProbability); + return saveRandomRecords(useAsync, hook, numRecords, recordGenerator); + } + + protected List> saveRandomRecords(final boolean useAsync, + @Nonnull final RecordMetaDataHook hook, + final int numRecords, + @Nonnull Function recordGenerator) throws Exception { if (useAsync) { return asyncBatch(hook, numRecords, 100, recNo -> recordStore.saveRecordAsync(recordGenerator.apply(recNo))); @@ -391,7 +398,7 @@ protected static Map> groupAndSortByDistances(@Nonnull final @Nonnull final Metric metric) { return storedRecords.stream() .map(storedRecord -> { - final VectorRecord vectorRecord = (VectorRecord)storedRecord.getRecord(); + final VectorRecord vectorRecord = VectorRecord.newBuilder().mergeFrom(storedRecord.getRecord()).build(); final RealVector storedVector = RealVector.fromBytes(vectorRecord.getVectorData().toByteArray()); return new NodeReferenceWithDistance(Tuple.from(vectorRecord.getRecNo()),