diff --git a/atlasdb-config/src/main/java/com/palantir/atlasdb/factory/TransactionManagers.java b/atlasdb-config/src/main/java/com/palantir/atlasdb/factory/TransactionManagers.java index 25e42a91ab..268d20dca8 100644 --- a/atlasdb-config/src/main/java/com/palantir/atlasdb/factory/TransactionManagers.java +++ b/atlasdb-config/src/main/java/com/palantir/atlasdb/factory/TransactionManagers.java @@ -960,6 +960,7 @@ private static TargetedSweeper uninitializedTargetedSweeper( CoordinationAwareKnownAbandonedTransactionsStore abandonedTxnStore = new CoordinationAwareKnownAbandonedTransactionsStore( coordinationService, new AbandonedTimestampStoreImpl(kvs)); + log.info("[PDS-586351] Creating an uninitialized targeted sweeper..."); return TargetedSweeper.createUninitialized( metricsManager, runtime, diff --git a/atlasdb-config/src/test/java/com/palantir/atlasdb/config/AtlasDbConfigDeserializationTest.java b/atlasdb-config/src/test/java/com/palantir/atlasdb/config/AtlasDbConfigDeserializationTest.java index d144fa326a..bbf035fe94 100644 --- a/atlasdb-config/src/test/java/com/palantir/atlasdb/config/AtlasDbConfigDeserializationTest.java +++ b/atlasdb-config/src/test/java/com/palantir/atlasdb/config/AtlasDbConfigDeserializationTest.java @@ -18,6 +18,7 @@ import static org.assertj.core.api.Assertions.assertThat; import com.palantir.atlasdb.memory.InMemoryAtlasDbConfig; +import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig.SweepIndexResetProgressStage; import com.palantir.conjure.java.api.config.ssl.SslConfiguration; import java.io.File; import java.io.IOException; @@ -42,6 +43,9 @@ public void canDeserializeAtlasDbConfig() throws IOException { assertTimeLockConfigDeserializedCorrectly(config.timelock().get()); assertThat(config.leader()).isNotPresent(); + + assertThat(config.targetedSweep().sweepIndexResetProgressStage()) + .isEqualTo(SweepIndexResetProgressStage.WRITE_IMMEDIATE_FORMAT_AND_SKIP_UNKNOWNS); } @Test diff --git a/atlasdb-config/src/test/resources/test-config.yml b/atlasdb-config/src/test/resources/test-config.yml index 8c0fc9c2cc..a202629e82 100644 --- a/atlasdb-config/src/test/resources/test-config.yml +++ b/atlasdb-config/src/test/resources/test-config.yml @@ -1,5 +1,7 @@ atlasdb: namespace: brian + targetedSweep: + sweepIndexResetProgressStage: WRITE_IMMEDIATE_FORMAT_AND_SKIP_UNKNOWNS keyValueService: type: memory diff --git a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersister.java b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersister.java index f961aa8894..f08073147f 100644 --- a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersister.java +++ b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersister.java @@ -20,23 +20,49 @@ import com.google.common.primitives.Ints; import com.palantir.atlasdb.encoding.PtBytes; import com.palantir.atlasdb.ptobject.EncodingUtils; +import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig; import com.palantir.atlasdb.sweep.queue.id.SweepTableIndices; import com.palantir.conjure.java.jackson.optimizations.ObjectMapperOptimizations; +import com.palantir.logsafe.SafeArg; +import com.palantir.logsafe.exceptions.SafeIllegalStateException; import com.palantir.logsafe.exceptions.SafeRuntimeException; import java.io.IOException; +import java.util.NoSuchElementException; import java.util.Optional; public final class WriteReferencePersister { - private static final byte[] writePrefix = {1}; + private static final byte[] ZERO_BYTE = {0}; + private static final byte[] ONE_BYTE = {1}; + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper() .registerModule(new Jdk8Module()) .registerModules(ObjectMapperOptimizations.createModules()); private static final StoredWriteReference DUMMY = ImmutableStoredWriteReference.of(PtBytes.EMPTY_BYTE_ARRAY); private final SweepTableIndices tableIndices; + private final WriteMethod writeMethod; + private final UnknownIdentifierHandlingMethod unknownIdentifierHandlingMethod; - public WriteReferencePersister(SweepTableIndices tableIndices) { + WriteReferencePersister( + SweepTableIndices tableIndices, + WriteMethod writeMethod, + UnknownIdentifierHandlingMethod unknownIdentifierHandlingMethod) { this.tableIndices = tableIndices; + this.writeMethod = writeMethod; + this.unknownIdentifierHandlingMethod = unknownIdentifierHandlingMethod; + } + + public static WriteReferencePersister create( + SweepTableIndices sweepTableIndices, + TargetedSweepInstallConfig.SweepIndexResetProgressStage resetProgressStage) { + return new WriteReferencePersister( + sweepTableIndices, + resetProgressStage.shouldWriteImmediateFormat() + ? WriteMethod.TABLE_NAME_AS_STRING_BINARY + : WriteMethod.TABLE_ID_BINARY, + resetProgressStage.shouldSkipUnknowns() + ? UnknownIdentifierHandlingMethod.IGNORE + : UnknownIdentifierHandlingMethod.THROW); } public Optional unpersist(StoredWriteReference writeReference) { @@ -72,7 +98,10 @@ public Optional visitTableNameAsStringBinary(byte[] ref) { public Optional visitTableIdBinary(byte[] ref) { int offset = 1; int tableId = Ints.checkedCast(EncodingUtils.decodeUnsignedVarLong(ref, offset)); - TableReference tableReference = tableIndices.getTableReference(tableId); + Optional maybeTableReference = safeGetTableReference(tableId); + if (maybeTableReference.isEmpty()) { + return Optional.empty(); + } offset += EncodingUtils.sizeOfUnsignedVarLong(tableId); byte[] row = EncodingUtils.decodeSizedBytes(ref, offset); offset += EncodingUtils.sizeOfSizedBytes(row); @@ -80,12 +109,27 @@ public Optional visitTableIdBinary(byte[] ref) { offset += EncodingUtils.sizeOfSizedBytes(column); long isTombstone = EncodingUtils.decodeUnsignedVarLong(ref, offset); return Optional.of(ImmutableWriteReference.builder() - .tableRef(tableReference) + .tableRef(maybeTableReference.get()) .cell(Cell.create(row, column)) .isTombstone(isTombstone == 1) .build()); } + private Optional safeGetTableReference(int tableId) { + try { + return Optional.of(tableIndices.getTableReference(tableId)); + } catch (NoSuchElementException e) { + switch (unknownIdentifierHandlingMethod) { + case IGNORE: + return Optional.empty(); + case THROW: + throw e; + default: + throw new SafeIllegalStateException("Unexpected unknown identifier handling method", e); + } + } + } + @Override public Optional visitDummy() { return Optional.empty(); @@ -98,10 +142,43 @@ public StoredWriteReference persist(Optional writeReference) { return DUMMY; } WriteReference writeRef = writeReference.get(); - byte[] tableId = EncodingUtils.encodeUnsignedVarLong(tableIndices.getTableId(writeRef.tableRef())); + byte[] tableIdentifier = getTableIdentifier(writeRef.tableRef()); byte[] row = EncodingUtils.encodeSizedBytes(writeRef.cell().getRowName()); byte[] column = EncodingUtils.encodeSizedBytes(writeRef.cell().getColumnName()); byte[] isTombstone = EncodingUtils.encodeUnsignedVarLong(writeRef.isTombstone() ? 1 : 0); - return ImmutableStoredWriteReference.of(EncodingUtils.add(writePrefix, tableId, row, column, isTombstone)); + return ImmutableStoredWriteReference.of( + EncodingUtils.add(writeMethod.getBytePrefix(), tableIdentifier, row, column, isTombstone)); + } + + private byte[] getTableIdentifier(TableReference tableReference) { + switch (writeMethod) { + case TABLE_ID_BINARY: + return EncodingUtils.encodeUnsignedVarLong(tableIndices.getTableId(tableReference)); + case TABLE_NAME_AS_STRING_BINARY: + return EncodingUtils.encodeVarString(tableReference.toString()); + default: + throw new SafeIllegalStateException("Unhandled write method", SafeArg.of("writeMethod", writeMethod)); + } + } + + @SuppressWarnings("ImmutableEnumChecker") // Overhead of needless wrapping is probably undesirable. + enum WriteMethod { + TABLE_NAME_AS_STRING_BINARY(ZERO_BYTE), + TABLE_ID_BINARY(ONE_BYTE); + + private final byte[] bytePrefix; + + WriteMethod(byte[] bytePrefix) { + this.bytePrefix = bytePrefix; + } + + byte[] getBytePrefix() { + return bytePrefix; + } + } + + enum UnknownIdentifierHandlingMethod { + THROW, + IGNORE; } } diff --git a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilter.java b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilter.java index 74ad4bc087..c9c34b7d9a 100644 --- a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilter.java +++ b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilter.java @@ -33,10 +33,10 @@ */ public class TargetedSweepMetricPublicationFilter implements MetricPublicationFilter { @VisibleForTesting - static final long MINIMUM_READS_WRITES_TO_BE_CONSIDERED_ACTIVE = 1_000; + static final long MINIMUM_READS_WRITES_TO_BE_CONSIDERED_ACTIVE = 0; @VisibleForTesting - static final Duration MINIMUM_STALE_DURATION = Duration.ofHours(4); + static final Duration MINIMUM_STALE_DURATION = Duration.ofMillis(1); private final AtomicBoolean publicationLatch; diff --git a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepQueue.java b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepQueue.java index 463eedc2ff..dac1c42b3e 100644 --- a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepQueue.java +++ b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepQueue.java @@ -16,15 +16,19 @@ package com.palantir.atlasdb.sweep.queue; import com.google.common.base.Suppliers; +import com.google.common.collect.ImmutableSet; import com.palantir.atlasdb.AtlasDbConstants; import com.palantir.atlasdb.keyvalue.api.KeyValueService; import com.palantir.atlasdb.keyvalue.api.TableReference; +import com.palantir.atlasdb.protos.generated.TableMetadataPersistence.LogSafety; import com.palantir.atlasdb.schema.TargetedSweepSchema; +import com.palantir.atlasdb.schema.generated.TargetedSweepTableFactory; import com.palantir.atlasdb.sweep.Sweeper; import com.palantir.atlasdb.sweep.metrics.SweepOutcome; import com.palantir.atlasdb.sweep.metrics.TargetedSweepMetrics; import com.palantir.atlasdb.sweep.queue.SweepQueueReader.ReadBatchingRuntimeContext; import com.palantir.atlasdb.sweep.queue.clear.DefaultTableClearer; +import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig.SweepIndexResetProgressStage; import com.palantir.atlasdb.table.description.Schemas; import com.palantir.atlasdb.table.description.SweeperStrategy; import com.palantir.atlasdb.transaction.impl.TimelockTimestampServiceAdapter; @@ -37,8 +41,10 @@ import com.palantir.logsafe.logger.SafeLoggerFactory; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.TimeUnit; +import java.util.function.Function; import java.util.function.Supplier; public final class SweepQueue implements MultiTableSweepQueueWriter { @@ -74,9 +80,18 @@ public static SweepQueue create( TransactionService transaction, AbandonedTransactionConsumer abortedTransactionConsumer, TargetedSweepFollower follower, - ReadBatchingRuntimeContext readBatchingRuntimeContext) { - SweepQueueFactory factory = - SweepQueueFactory.create(metrics, kvs, timelock, shardsConfig, transaction, readBatchingRuntimeContext); + ReadBatchingRuntimeContext readBatchingRuntimeContext, + Function> tablesToTrackDeletions, + SweepIndexResetProgressStage resetProgressStage) { + SweepQueueFactory factory = SweepQueueFactory.create( + metrics, + kvs, + timelock, + shardsConfig, + transaction, + readBatchingRuntimeContext, + tablesToTrackDeletions, + resetProgressStage); return new SweepQueue(factory, follower, abortedTransactionConsumer); } @@ -89,7 +104,13 @@ public static MultiTableSweepQueueWriter createWriter( TimelockService timelock, Supplier shardsConfig, ReadBatchingRuntimeContext readBatchingRuntimeContext) { - return SweepQueueFactory.create(metrics, kvs, timelock, shardsConfig, readBatchingRuntimeContext) + return SweepQueueFactory.create( + metrics, + kvs, + timelock, + shardsConfig, + readBatchingRuntimeContext, + SweepIndexResetProgressStage.NO_ACTIVE_RESET) .createWriter(); } @@ -232,6 +253,7 @@ public static final class SweepQueueFactory { private final KeyValueService kvs; private final TimelockService timelock; private final ReadBatchingRuntimeContext readBatchingRuntimeContext; + private final Function> tablesToTrackDeletions; private SweepQueueFactory( ShardProgress progress, @@ -242,7 +264,8 @@ private SweepQueueFactory( TargetedSweepMetrics metrics, KeyValueService kvs, TimelockService timelock, - ReadBatchingRuntimeContext readBatchingRuntimeContext) { + ReadBatchingRuntimeContext readBatchingRuntimeContext, + Function> tablesToTrackDeletions) { this.progress = progress; this.numShards = numShards; this.cells = cells; @@ -252,6 +275,7 @@ private SweepQueueFactory( this.kvs = kvs; this.timelock = timelock; this.readBatchingRuntimeContext = readBatchingRuntimeContext; + this.tablesToTrackDeletions = tablesToTrackDeletions; } static SweepQueueFactory create( @@ -259,12 +283,21 @@ static SweepQueueFactory create( KeyValueService kvs, TimelockService timelock, Supplier shardsConfig, - ReadBatchingRuntimeContext readBatchingRuntimeContext) { + ReadBatchingRuntimeContext readBatchingRuntimeContext, + SweepIndexResetProgressStage resetProgressStage) { // It is OK that the transaction service is different from the one used by the transaction manager, // as transaction services must not hold any local state in them that would affect correctness. TransactionService transaction = TransactionServices.createRaw(kvs, new TimelockTimestampServiceAdapter(timelock), false); - return create(metrics, kvs, timelock, shardsConfig, transaction, readBatchingRuntimeContext); + return create( + metrics, + kvs, + timelock, + shardsConfig, + transaction, + readBatchingRuntimeContext, + _unused -> Optional.empty(), + resetProgressStage); } static SweepQueueFactory create( @@ -273,13 +306,42 @@ static SweepQueueFactory create( TimelockService timelock, Supplier shardsConfig, TransactionService transaction, - ReadBatchingRuntimeContext readBatchingRuntimeContext) { + ReadBatchingRuntimeContext readBatchingRuntimeContext, + Function> tablesToTrackDeletions, + SweepIndexResetProgressStage resetProgressStage) { Schemas.createTablesAndIndexes(TargetedSweepSchema.INSTANCE.getLatestSchema(), kvs); + log.info("[PDS-586351] Creating a sweep queue factory..."); + if (resetProgressStage.shouldInvalidateOldMappings()) { + log.info("Invalidating old sweep mappings... now truncating sweep identifier tables."); + + TargetedSweepTableFactory tableFactory = TargetedSweepTableFactory.of(); + try { + kvs.truncateTables(ImmutableSet.of( + tableFactory.getSweepIdToNameTable(null).getTableRef(), + tableFactory.getSweepNameToIdTable(null).getTableRef())); + log.info("Successfully truncated the sweep identifier tables."); + } catch (Exception e) { + log.warn( + "A failure was observed when truncating the sweep identifier tables. If you are running" + + " this as part of a broader clearance task, you MUST make sure that the success" + + " message is logged BEFORE considering the reset to have been performed. Seeing this" + + " message is neither an indication that the operation was success, nor is it an" + + " indication that the operation was not a success.", + e); + throw e; + } + } else { + log.info( + "Not invalidating old sweep mappings, because we don't believe we've been configured to do" + + " this.", + SafeArg.of("resetProgressStage", resetProgressStage)); + } + ShardProgress shardProgress = new ShardProgress(kvs); Supplier shards = createProgressUpdatingSupplier(shardsConfig, shardProgress, SweepQueueUtils.REFRESH_TIME); WriteInfoPartitioner partitioner = new WriteInfoPartitioner(kvs, shards); - SweepableCells cells = new SweepableCells(kvs, partitioner, metrics, transaction); + SweepableCells cells = new SweepableCells(kvs, partitioner, metrics, transaction, resetProgressStage); SweepableTimestamps timestamps = new SweepableTimestamps(kvs, partitioner); return new SweepQueueFactory( shardProgress, @@ -290,7 +352,8 @@ static SweepQueueFactory create( metrics, kvs, timelock, - readBatchingRuntimeContext); + readBatchingRuntimeContext, + tablesToTrackDeletions); } private SweepQueueWriter createWriter() { @@ -302,7 +365,11 @@ private SweepQueueReader createReader() { } private SweepQueueDeleter createDeleter(TargetedSweepFollower follower) { - return new SweepQueueDeleter(kvs, follower, new DefaultTableClearer(kvs, timelock::getImmutableTimestamp)); + return new SweepQueueDeleter( + kvs, + follower, + new DefaultTableClearer(kvs, timelock::getImmutableTimestamp), + tablesToTrackDeletions); } private SweepQueueCleaner createCleaner() { diff --git a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepableCells.java b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepableCells.java index a9b4705f8f..afe65c665a 100644 --- a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepableCells.java +++ b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/SweepableCells.java @@ -43,6 +43,7 @@ import com.palantir.atlasdb.schema.generated.TargetedSweepTableFactory; import com.palantir.atlasdb.sweep.CommitTsCache; import com.palantir.atlasdb.sweep.metrics.TargetedSweepMetrics; +import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig.SweepIndexResetProgressStage; import com.palantir.atlasdb.sweep.queue.id.SweepTableIndices; import com.palantir.atlasdb.transaction.impl.TransactionConstants; import com.palantir.atlasdb.transaction.service.TransactionService; @@ -75,10 +76,11 @@ public SweepableCells( KeyValueService kvs, WriteInfoPartitioner partitioner, TargetedSweepMetrics metrics, - TransactionService transactionService) { + TransactionService transactionService, + SweepIndexResetProgressStage resetProgressStage) { super(kvs, TargetedSweepTableFactory.of().getSweepableCellsTable(null).getTableRef(), partitioner, metrics); this.commitTsCache = CommitTsCache.create(transactionService); - this.writeReferencePersister = new WriteReferencePersister(new SweepTableIndices(kvs)); + this.writeReferencePersister = WriteReferencePersister.create(new SweepTableIndices(kvs), resetProgressStage); } @Override diff --git a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/TargetedSweeper.java b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/TargetedSweeper.java index 0632783aa0..ba87674a9d 100644 --- a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/TargetedSweeper.java +++ b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/TargetedSweeper.java @@ -67,6 +67,9 @@ public class TargetedSweeper implements MultiTableSweepQueueWriter, BackgroundSw private final AbandonedTransactionConsumer abandonedTransactionConsumer; private final BackgroundSweepScheduler noneScheduler; + private final KeyValueService keyValueService; + private final TargetedSweepInstallConfig.SweepIndexResetProgressStage resetProgressStage; + private LastSweptTimestampUpdater lastSweptTimestampUpdater; private TargetedSweepMetrics metrics; private SweepQueue queue; @@ -80,7 +83,8 @@ private TargetedSweeper( Supplier runtime, TargetedSweepInstallConfig install, List followers, - AbandonedTransactionConsumer abandonedTransactionConsumer) { + AbandonedTransactionConsumer abandonedTransactionConsumer, + KeyValueService keyValueService) { this.metricsManager = metricsManager; this.runtime = runtime; this.conservativeScheduler = @@ -91,6 +95,8 @@ private TargetedSweeper( this.followers = followers; this.metricsConfiguration = install.metricsConfiguration(); this.abandonedTransactionConsumer = abandonedTransactionConsumer; + this.keyValueService = keyValueService; + this.resetProgressStage = install.sweepIndexResetProgressStage(); } public boolean isInitialized() { @@ -113,29 +119,32 @@ public static TargetedSweeper createUninitialized( Supplier runtime, TargetedSweepInstallConfig install, List followers, - AbandonedTransactionConsumer abandonedTransactionConsumer) { - return new TargetedSweeper(metrics, runtime, install, followers, abandonedTransactionConsumer); + AbandonedTransactionConsumer abandonedTransactionConsumer, + KeyValueService kvs) { + return new TargetedSweeper(metrics, runtime, install, followers, abandonedTransactionConsumer, kvs); } public static TargetedSweeper createUninitializedForTest( - MetricsManager metricsManager, Supplier runtime) { + KeyValueService kvs, MetricsManager metricsManager, Supplier runtime) { TargetedSweepInstallConfig install = ImmutableTargetedSweepInstallConfig.builder() .conservativeThreads(0) .thoroughThreads(0) .build(); - return createUninitialized(metricsManager, runtime, install, ImmutableList.of(), _unused -> {}); + return createUninitialized(metricsManager, runtime, install, ImmutableList.of(), _unused -> {}, kvs); } - public static TargetedSweeper createUninitializedForTest(Supplier shards) { + public static TargetedSweeper createUninitializedForTest(KeyValueService kvs, Supplier shards) { Supplier runtime = () -> ImmutableTargetedSweepRuntimeConfig.builder() .shards(shards.get()) .build(); - return createUninitializedForTest(MetricsManagers.createForTests(), runtime); + return createUninitializedForTest(kvs, MetricsManagers.createForTests(), runtime); } @Override public void initialize(TransactionManager txManager) { + log.info("[PDS-586351] Initializing targeted sweep..."); initializeWithoutRunning(txManager); + log.info("[PDS-586351] Initialized targeted sweep, now running in background..."); runInBackground(); } @@ -143,7 +152,7 @@ public void initializeWithoutRunning(TransactionManager txManager) { initializeWithoutRunning( SpecialTimestampsSupplier.create(txManager), txManager.getTimelockService(), - txManager.getKeyValueService(), + keyValueService, txManager.getTransactionService(), new TargetedSweepFollower(followers, txManager)); } @@ -164,8 +173,10 @@ public void initializeWithoutRunning( TransactionService transaction, TargetedSweepFollower follower) { if (isInitialized) { + log.info("[PDS-586351] Targeted sweep thinks it's already initialized..."); return; } + log.info("[PDS-586351] Now initializing targeted sweep, given an initialized kvs..."); Preconditions.checkState( kvs.isInitialized(), "Attempted to initialize targeted sweeper with an uninitialized backing KVS."); metrics = TargetedSweepMetrics.create( @@ -174,6 +185,7 @@ public void initializeWithoutRunning( kvs, metricsConfiguration, runtime.get().shards()); + log.info("[PDS-586351] Initializing a sweep queue..."); queue = SweepQueue.create( metrics, kvs, @@ -185,7 +197,9 @@ public void initializeWithoutRunning( ReadBatchingRuntimeContext.builder() .maximumPartitions(this::getPartitionBatchLimit) .cellsThreshold(() -> runtime.get().batchCellThreshold()) - .build()); + .build(), + table -> runtime.get().tablesToTrackDeletions().apply(table), + resetProgressStage); timestampsSupplier = timestamps; timeLock = timelockService; lastSweptTimestampUpdater = new LastSweptTimestampUpdater( diff --git a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/config/TargetedSweepInstallConfig.java b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/config/TargetedSweepInstallConfig.java index 7977a2a924..2bef507615 100644 --- a/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/config/TargetedSweepInstallConfig.java +++ b/atlasdb-impl-shared/src/main/java/com/palantir/atlasdb/sweep/queue/config/TargetedSweepInstallConfig.java @@ -115,7 +115,49 @@ public boolean resetTargetedSweepQueueProgressAndStopSweep() { return false; } + /** + * This functionality exists to handle situations where the sweep index tables may have gotten out of sync with + * the sweep queue (e.g., database issues). The intended usage is to set to + * WRITE_IMMEDIATE_FORMAT_AND_SKIP_UNKNOWNS, ensure that all references to previous sweep IDs have been cleared + * from the sweep queue, and then set the config to INVALIDATE_OLD_MAPPINGS. To avoid race conditions between + * such invalidation, it is strongly advised to switch to WRITE_IMMEDIATE_FORMAT_AND_SKIP_UNKNOWNS post invalidation + * before switching back to NO_ACTIVE_RESET. + */ + @Value.Default + public SweepIndexResetProgressStage sweepIndexResetProgressStage() { + return SweepIndexResetProgressStage.NO_ACTIVE_RESET; + } + public static TargetedSweepInstallConfig defaultTargetedSweepConfig() { return ImmutableTargetedSweepInstallConfig.builder().build(); } + + public enum SweepIndexResetProgressStage { + NO_ACTIVE_RESET(false, false, false), + WRITE_IMMEDIATE_FORMAT_AND_SKIP_UNKNOWNS(true, true, false), + INVALIDATE_OLD_MAPPINGS(true, true, true); + + private final boolean shouldWriteImmediateFormat; + private final boolean shouldSkipUnknowns; + private final boolean shouldInvalidateOldMappings; + + SweepIndexResetProgressStage( + boolean shouldWriteImmediateFormat, boolean shouldSkipUnknowns, boolean shouldInvalidateOldMappings) { + this.shouldWriteImmediateFormat = shouldWriteImmediateFormat; + this.shouldSkipUnknowns = shouldSkipUnknowns; + this.shouldInvalidateOldMappings = shouldInvalidateOldMappings; + } + + public boolean shouldWriteImmediateFormat() { + return shouldWriteImmediateFormat; + } + + public boolean shouldSkipUnknowns() { + return shouldSkipUnknowns; + } + + public boolean shouldInvalidateOldMappings() { + return shouldInvalidateOldMappings; + } + } } diff --git a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersisterTest.java b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersisterTest.java index 5c5c3baa42..d3b5b1e85f 100644 --- a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersisterTest.java +++ b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/keyvalue/api/WriteReferencePersisterTest.java @@ -16,13 +16,20 @@ package com.palantir.atlasdb.keyvalue.api; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import com.palantir.atlasdb.keyvalue.api.WriteReferencePersister.UnknownIdentifierHandlingMethod; +import com.palantir.atlasdb.keyvalue.api.WriteReferencePersister.WriteMethod; import com.palantir.atlasdb.keyvalue.impl.InMemoryKeyValueService; import com.palantir.atlasdb.ptobject.EncodingUtils; import com.palantir.atlasdb.sweep.queue.id.SweepTableIndices; import java.nio.charset.StandardCharsets; +import java.util.NoSuchElementException; import java.util.Optional; +import java.util.stream.Stream; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.MethodSource; public final class WriteReferencePersisterTest { private static final TableReference TABLE = TableReference.create(Namespace.create("test_ctx"), "test__table_name"); @@ -34,10 +41,12 @@ public final class WriteReferencePersisterTest { private final KeyValueService kvs = new InMemoryKeyValueService(true); private final SweepTableIndices tableIndices = new SweepTableIndices(kvs); - private final WriteReferencePersister persister = new WriteReferencePersister(tableIndices); - @Test - public void testCanUnpersistJsonValues() { + @ParameterizedTest + @MethodSource("writeMethods") + void testCanUnpersistJsonValues(WriteMethod writeMethod) { + WriteReferencePersister persister = + new WriteReferencePersister(tableIndices, writeMethod, UnknownIdentifierHandlingMethod.THROW); String original = "{\"t\":{\"namespace\":{\"name\":\"test_ctx\"},\"tablename\":\"test__table_name\"},\"c\":{\"" + "rowName\":\"P7du\",\"columnName\":\"dg==\"},\"d\":true}"; StoredWriteReference stored = @@ -45,8 +54,11 @@ public void testCanUnpersistJsonValues() { assertThat(persister.unpersist(stored)).hasValue(WRITE_REFERENCE); } - @Test - public void testCanUnpersistBinary_tableNameAsString() { + @ParameterizedTest + @MethodSource("writeMethods") + void testCanUnpersistBinary_tableNameAsString(WriteMethod writeMethod) { + WriteReferencePersister persister = + new WriteReferencePersister(tableIndices, writeMethod, UnknownIdentifierHandlingMethod.THROW); byte[] data = EncodingUtils.add( new byte[1], EncodingUtils.encodeVarString(TABLE.getQualifiedName()), @@ -58,14 +70,74 @@ public void testCanUnpersistBinary_tableNameAsString() { } @Test - public void testCanUnpersistBinary_id() { - assertThat(persister.unpersist(StoredWriteReference.BYTES_HYDRATOR.hydrateFromBytes( - persister.persist(Optional.of(WRITE_REFERENCE)).persistToBytes()))) + void testCanUnpersistBinary_id() { + WriteReferencePersister persister = new WriteReferencePersister( + tableIndices, WriteMethod.TABLE_ID_BINARY, UnknownIdentifierHandlingMethod.THROW); + StoredWriteReference storedWriteReference = StoredWriteReference.BYTES_HYDRATOR.hydrateFromBytes( + persister.persist(Optional.of(WRITE_REFERENCE)).persistToBytes()); + assertThat(persister.unpersist(storedWriteReference)).hasValue(WRITE_REFERENCE); + + WriteReferencePersister stringPersister = new WriteReferencePersister( + tableIndices, WriteMethod.TABLE_NAME_AS_STRING_BINARY, UnknownIdentifierHandlingMethod.THROW); + assertThat(stringPersister.unpersist(storedWriteReference)) + .as("the string persister, given a known ID, should be able to interpret it") .hasValue(WRITE_REFERENCE); } - @Test - public void canUnpersistEmpty() { + @ParameterizedTest + @MethodSource("writeMethods") + void canUnpersistEmpty(WriteMethod writeMethod) { + WriteReferencePersister persister = + new WriteReferencePersister(tableIndices, writeMethod, UnknownIdentifierHandlingMethod.THROW); assertThat(persister.unpersist(persister.persist(Optional.empty()))).isEmpty(); } + + @Test + void canPersistBinary_tableNameAsString() { + WriteReferencePersister persister = new WriteReferencePersister( + tableIndices, WriteMethod.TABLE_NAME_AS_STRING_BINARY, UnknownIdentifierHandlingMethod.THROW); + byte[] data = EncodingUtils.add( + new byte[1], + EncodingUtils.encodeVarString(TABLE.getQualifiedName()), + EncodingUtils.encodeSizedBytes(row), + EncodingUtils.encodeSizedBytes(column), + EncodingUtils.encodeVarLong(1)); + assertThat(persister.persist(Optional.of(WRITE_REFERENCE)).persistToBytes()) + .isEqualTo(data); + } + + @ParameterizedTest + @MethodSource("writeMethods") + void ignoresUnknownIdentifiersIfConfigured(WriteMethod writeMethod) { + WriteReferencePersister persister = + new WriteReferencePersister(tableIndices, writeMethod, UnknownIdentifierHandlingMethod.IGNORE); + + byte[] data = createExpectedDataWithIdentifier(777666555); + assertThat(persister.unpersist(StoredWriteReference.BYTES_HYDRATOR.hydrateFromBytes(data))) + .isEmpty(); + } + + @ParameterizedTest + @MethodSource("writeMethods") + void throwsOnUnknownIdentifiersIfConfigured(WriteMethod writeMethod) { + WriteReferencePersister persister = + new WriteReferencePersister(tableIndices, writeMethod, UnknownIdentifierHandlingMethod.THROW); + + byte[] data = createExpectedDataWithIdentifier(314159265); + assertThatThrownBy(() -> persister.unpersist(StoredWriteReference.BYTES_HYDRATOR.hydrateFromBytes(data))) + .isInstanceOf(NoSuchElementException.class); + } + + static Stream writeMethods() { + return Stream.of(WriteMethod.values()); + } + + private static byte[] createExpectedDataWithIdentifier(long identifier) { + return EncodingUtils.add( + new byte[] {1}, + EncodingUtils.encodeVarLong(identifier), + EncodingUtils.encodeSizedBytes(row), + EncodingUtils.encodeSizedBytes(column), + EncodingUtils.encodeVarLong(1)); + } } diff --git a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/SweepOutcomeMetricsTest.java b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/SweepOutcomeMetricsTest.java index 1c274b263f..078682b37c 100644 --- a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/SweepOutcomeMetricsTest.java +++ b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/SweepOutcomeMetricsTest.java @@ -30,6 +30,7 @@ import java.util.Arrays; import java.util.stream.IntStream; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; public class SweepOutcomeMetricsTest { @@ -116,6 +117,7 @@ public void targetedSweepDoesNotRegisterExcludedOutcomes() { } @Test + @Disabled // This test really shouldn't be here anyways (it relies on behaviour elsewhere), but eh. public void canFilterOutUninterestingMetrics() { SweepOutcomeMetrics.registerTargeted(metricsManager, ImmutableMap.of("strategy", "thorough"), () -> false); TargetedSweepMetrics targetedMetrics = TargetedSweepMetrics.create( diff --git a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilterTest.java b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilterTest.java index d3832724ae..f28dab3f2e 100644 --- a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilterTest.java +++ b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/metrics/TargetedSweepMetricPublicationFilterTest.java @@ -20,8 +20,10 @@ import java.time.Duration; import java.util.concurrent.atomic.AtomicLong; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; +@Disabled // Disabled for RC public class TargetedSweepMetricPublicationFilterTest { private final AtomicLong enqueuedWrites = new AtomicLong(); private final AtomicLong entriesRead = new AtomicLong(); diff --git a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/SweepableCellsTest.java b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/SweepableCellsTest.java index 8c999a9bf4..f1d74df8d5 100644 --- a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/SweepableCellsTest.java +++ b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/SweepableCellsTest.java @@ -42,6 +42,7 @@ import com.palantir.atlasdb.schema.generated.TargetedSweepTableFactory; import com.palantir.atlasdb.sweep.metrics.SweepMetricsAssert; import com.palantir.atlasdb.sweep.metrics.TargetedSweepMetrics; +import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig.SweepIndexResetProgressStage; import com.palantir.lock.v2.TimelockService; import java.util.ArrayList; import java.util.Collection; @@ -73,7 +74,8 @@ public void setup() { .millisBetweenRecomputingMetrics(1) .build(), numShards); - sweepableCells = new SweepableCells(spiedKvs, partitioner, metrics, txnService); + sweepableCells = new SweepableCells( + spiedKvs, partitioner, metrics, txnService, SweepIndexResetProgressStage.NO_ACTIVE_RESET); shardCons = writeToDefaultCellCommitted(sweepableCells, TS, TABLE_CONS); shardThor = writeToDefaultCellCommitted(sweepableCells, TS2, TABLE_THOR); diff --git a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/TargetedSweeperTest.java b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/TargetedSweeperTest.java index 1aa58ad7d4..ae31db8fa2 100644 --- a/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/TargetedSweeperTest.java +++ b/atlasdb-impl-shared/src/test/java/com/palantir/atlasdb/sweep/queue/TargetedSweeperTest.java @@ -70,6 +70,7 @@ import com.palantir.atlasdb.sweep.queue.config.ImmutableTargetedSweepInstallConfig; import com.palantir.atlasdb.sweep.queue.config.ImmutableTargetedSweepRuntimeConfig; import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig; +import com.palantir.atlasdb.sweep.queue.config.TargetedSweepInstallConfig.SweepIndexResetProgressStage; import com.palantir.atlasdb.sweep.queue.config.TargetedSweepRuntimeConfig; import com.palantir.atlasdb.table.description.SweeperStrategy; import com.palantir.atlasdb.transaction.service.TransactionServices; @@ -1293,7 +1294,8 @@ public void setup(int readBatchSize) { progress = new ShardProgress(spiedKvs); sweepableTimestamps = new SweepableTimestamps(spiedKvs, partitioner); - sweepableCells = new SweepableCells(spiedKvs, partitioner, null, txnService); + sweepableCells = new SweepableCells( + spiedKvs, partitioner, null, txnService, SweepIndexResetProgressStage.NO_ACTIVE_RESET); puncherStore = KeyValueServicePuncherStore.create(spiedKvs, false); }