TargetedSweeper implements MultiTableSweepQueueWriter, BackgroundSweeper { @SuppressWarnings("checkstyle:RegexpMultiline") @VisibleForTesting public long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive) { assertInitialized(); return queue.sweepNextBatch(shardStrategy, maxTsExclusive); } private  TargetedSweeper(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); static TargetedSweeper createUninitialized(MetricsManager metrics,
            Supplier<TargetedSweepRuntimeConfig> runtime,
            TargetedSweepInstallConfig install,
            List<Follower> followers); static TargetedSweeper createUninitializedForTest(MetricsManager metricsManager,
            Supplier<TargetedSweepRuntimeConfig> runtime); static TargetedSweeper createUninitializedForTest(Supplier<Integer> shards); @Override void initialize(TransactionManager txManager); void initializeWithoutRunning(TransactionManager txManager); void initializeWithoutRunning(
            SpecialTimestampsSupplier timestamps,
            TimelockService timelockService,
            KeyValueService kvs,
            TransactionService transaction,
            TargetedSweepFollower follower); @Override void runInBackground(); @Override void enqueue(List<WriteInfo> writes); @SuppressWarnings("checkstyle:RegexpMultiline") // Suppress VisibleForTesting warning @VisibleForTesting long sweepNextBatch(ShardAndStrategy shardStrategy, long maxTsExclusive); @Override void close(); @Override void shutdown();  }
@Test public void conservativeSweepAddsSentinelAndLeavesSingleValue() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); assertReadAtTimestampReturnsNothing(TABLE_CONS, LOW_TS); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS); }
@Test public void sweepsThoroughMigrationAsConservative() { enqueueWriteCommitted(TABLE_THOR_MIGRATION, LOW_TS); assertReadAtTimestampReturnsNothing(TABLE_THOR_MIGRATION, LOW_TS); sweepNextBatch(ShardAndStrategy.conservative(THOR_MIGRATION_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_THOR_MIGRATION, LOW_TS); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_THOR_MIGRATION, LOW_TS); }
@Test public void sweepWithSingleEntryUpdatesMetrics() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(1); assertThat(metricsManager).hasSweepTimestampConservativeEqualTo( getSweepTsCons()); assertThat(metricsManager).hasLastSweptTimestampConservativeEqualTo( maxTsForFinePartition(0)); setTimelockTime(5_000L); punchTimeAtTimestamp(2_000, LOW_TS); assertThat(metricsManager).hasMillisSinceLastSweptConservativeEqualTo(5_000L - 2_000L); assertThat(metricsManager).hasTargetedOutcomeEqualTo(CONSERVATIVE, SweepOutcome.SUCCESS, 1L); assertThat(metricsManager).hasNotRegisteredTargetedOutcome(THOROUGH, SweepOutcome.SUCCESS); }
@Test public void sweepWithNoCandidatesBeforeSweepTimestampReportsNothingToSweep() { enqueueWriteCommitted(TABLE_CONS, getSweepTsCons()); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasTargetedOutcomeEqualTo(CONSERVATIVE, SweepOutcome.NOTHING_TO_SWEEP, 1L); assertThat(metricsManager).hasNotRegisteredTargetedOutcome(THOROUGH, SweepOutcome.NOTHING_TO_SWEEP); }
@Test public void thoroughSweepDoesNotAddSentinelAndLeavesSingleValue() { enqueueWriteCommitted(TABLE_THOR, LOW_TS); assertReadAtTimestampReturnsNothing(TABLE_THOR, LOW_TS); sweepNextBatch(ShardAndStrategy.thorough(THOR_SHARD)); assertReadAtTimestampReturnsNothing(TABLE_THOR, LOW_TS); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_THOR, LOW_TS); }
@Test public void thoroughSweepDeletesExistingSentinel() { spiedKvs.addGarbageCollectionSentinelValues(TABLE_THOR, ImmutableList.of(DEFAULT_CELL)); assertReadAtTimestampReturnsSentinel(TABLE_THOR, 0L); enqueueWriteCommitted(TABLE_THOR, 10L); sweepNextBatch(ShardAndStrategy.thorough(THOR_SHARD)); assertReadAtTimestampReturnsNothing(TABLE_THOR, 0L); }
@Test public void conservativeSweepDeletesLowerValue() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); enqueueWriteCommitted(TABLE_CONS, LOW_TS2); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS2); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + 1); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS2); }
@Test public void thoroughSweepDeletesLowerValue() { enqueueWriteCommitted(TABLE_THOR, LOW_TS); enqueueWriteCommitted(TABLE_THOR, LOW_TS2); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_THOR, LOW_TS); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_THOR, LOW_TS2); sweepNextBatch(ShardAndStrategy.thorough(THOR_SHARD)); assertReadAtTimestampReturnsNothing(TABLE_THOR, LOW_TS + 1); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_THOR, LOW_TS2); }
@Test public void conservativeSweepCallsFollower() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); enqueueWriteCommitted(TABLE_CONS, LOW_TS2); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); ArgumentCaptor<Set> captor = ArgumentCaptor.forClass(Set.class); verify(mockFollower, times(1)).run(eq(TABLE_CONS), captor.capture()); assertThat(Iterables.getOnlyElement(captor.getAllValues())).containsExactly(DEFAULT_CELL); }
@Test public void thoroughSweepCallsFollower() { enqueueWriteCommitted(TABLE_THOR, LOW_TS); enqueueWriteCommitted(TABLE_THOR, LOW_TS2); sweepNextBatch(ShardAndStrategy.thorough(THOR_SHARD)); ArgumentCaptor<Set> captor = ArgumentCaptor.forClass(Set.class); verify(mockFollower, times(1)).run(eq(TABLE_THOR), captor.capture()); assertThat(Iterables.getOnlyElement(captor.getAllValues())).containsExactly(DEFAULT_CELL); }
@Test public void conservativeSweepDeletesAllButLatestWithSingleDeleteAllTimestamps() { long lastWriteTs = 5000; for (long i = 1; i <= lastWriteTs; i++) { enqueueWriteCommitted(TABLE_CONS, i); } sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, lastWriteTs); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, lastWriteTs); verify(spiedKvs, times(1)).deleteAllTimestamps(any(TableReference.class), anyMap()); }
@Test public void thoroughSweepDeletesAllButLatestWithSingleDeleteAllTimestampsIncludingSentinels() { long lastWriteTs = 5000; for (long i = 1; i <= lastWriteTs; i++) { enqueueWriteCommitted(TABLE_THOR, i); } sweepNextBatch(ShardAndStrategy.thorough(THOR_SHARD)); assertReadAtTimestampReturnsNothing(TABLE_THOR, lastWriteTs); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_THOR, lastWriteTs); verify(spiedKvs, times(1)).deleteAllTimestamps(any(TableReference.class), anyMap()); }
@Test public void sweepsOnlyThePrescribedNumberOfBatchesAtATime() { for (int partition = 0; partition <= readBatchSize; partition++) { enqueueWriteCommitted(TABLE_CONS, LOW_TS + SweepQueueUtils.minTsForFinePartition(partition)); enqueueWriteCommitted(TABLE_CONS, LOW_TS + SweepQueueUtils.minTsForFinePartition(partition) + 1); } sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel( TABLE_CONS, LOW_TS + SweepQueueUtils.minTsForFinePartition(readBatchSize - 1)); assertTestValueEnqueuedAtGivenTimestampStillPresent( TABLE_CONS, LOW_TS + SweepQueueUtils.minTsForFinePartition(readBatchSize - 1) + 1); assertTestValueEnqueuedAtGivenTimestampStillPresent( TABLE_CONS, LOW_TS + SweepQueueUtils.minTsForFinePartition(readBatchSize)); assertThat(metricsManager).containsEntriesReadInBatchConservative(readBatchSize * 2); assertThat(metricsManager).containsEntriesReadInBatchThorough(); assertThat(metricsManager).hasEntriesReadInBatchMeanConservativeEqualTo(readBatchSize * 2); assertThat(metricsManager).hasEntriesReadInBatchMeanThoroughEqualTo(0.0); }
@Test public void sweepDeletesWritesWhenTombstoneHasHigherTimestamp() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); enqueueTombstone(TABLE_CONS, LOW_TS2); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS); assertReadAtTimestampReturnsTombstoneAtTimestamp(TABLE_CONS, LOW_TS2 + 1, LOW_TS2); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + 1); assertReadAtTimestampReturnsTombstoneAtTimestamp(TABLE_CONS, LOW_TS2 + 1, LOW_TS2); }
@Test public void thoroughSweepDeletesTombstoneIfLatestWrite() { enqueueTombstone(TABLE_THOR, LOW_TS); enqueueTombstone(TABLE_THOR, LOW_TS2); assertReadAtTimestampReturnsTombstoneAtTimestamp(TABLE_THOR, LOW_TS + 1, LOW_TS); assertReadAtTimestampReturnsTombstoneAtTimestamp(TABLE_THOR, LOW_TS2 + 1, LOW_TS2); sweepNextBatch(ShardAndStrategy.thorough(THOR_SHARD)); assertReadAtTimestampReturnsNothing(TABLE_THOR, LOW_TS + 1); assertReadAtTimestampReturnsNothing(TABLE_THOR, LOW_TS2 + 1); }
@Test public void sweepDeletesTombstonesWhenWriteHasHigherTimestamp() { enqueueTombstone(TABLE_CONS, LOW_TS); enqueueWriteCommitted(TABLE_CONS, LOW_TS2); assertReadAtTimestampReturnsTombstoneAtTimestamp(TABLE_CONS, LOW_TS + 1, LOW_TS); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS2); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + 1); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS2); }
@Test public void sweepHandlesSequencesOfDeletesAndReadditionsInOneShot() { enqueueWriteCommitted(TABLE_CONS, LOW_TS); enqueueTombstone(TABLE_CONS, LOW_TS + 2); enqueueWriteCommitted(TABLE_CONS, LOW_TS + 4); enqueueTombstone(TABLE_CONS, LOW_TS + 6); enqueueWriteCommitted(TABLE_CONS, LOW_TS + 8); for (int partition = 0; partition < readBatchSize; partition++) { enqueueWriteCommitted(TABLE_CONS, minTsForFinePartition(partition + 1)); } sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); for (int i = 0; i < 10; i = i + 2) { assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + i); } if (readBatchSize == 1) { assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS + 8); } else { assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, minTsForFinePartition(readBatchSize)); } verify(spiedKvs, times(1)).deleteAllTimestamps(any(), any()); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(1); assertThat(metricsManager).hasLastSweptTimestampConservativeEqualTo( maxTsForFinePartition(readBatchSize - 1)); setTimelockTime(10_000L); punchTimeAtTimestamp(5_000L, LOW_TS + 8); assertThat(metricsManager).hasMillisSinceLastSweptConservativeEqualTo(10_000L - 5000L); }
@Test public void enableAutoTuningOverridesEffectivelySetsLargeReadBatchSize() { enableAutoTuning = true; for (int partition = 0; partition < readBatchSize * 5; partition++) { enqueueWriteCommitted(TABLE_CONS, minTsForFinePartition(partition + 1)); } sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); for (int partition = 0; partition < readBatchSize * 5 - 1; partition++) { assertReadAtTimestampReturnsSentinel(TABLE_CONS, minTsForFinePartition(partition + 1)); } assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, minTsForFinePartition(readBatchSize * 5)); verify(spiedKvs, times(1)).deleteAllTimestamps(any(), any()); }
@Test public void sweepProgressesAndSkipsEmptyFinePartitions() { setSweepTimestamp(minTsForFinePartition(2 * (2 * readBatchSize) + 2)); List<Integer> permittedPartitions = Lists.newArrayList(); for (int index = 0; index <= 2 * readBatchSize; index++) { int partitionToUse = 2 * index; long timestampToUse = minTsForFinePartition(partitionToUse) + LOW_TS; enqueueWriteCommitted(TABLE_CONS, timestampToUse); punchTimeAtTimestamp(100 * (index + 1), timestampToUse); permittedPartitions.add(partitionToUse); } int finalPartition = 2 * (2 * readBatchSize); long additionalTimestampForFinalPartition = minTsForFinePartition(finalPartition) + LOW_TS + 1; enqueueWriteCommitted(TABLE_CONS, additionalTimestampForFinalPartition); int finalValueWallClockTime = 100 * (2 * readBatchSize + 1) + 1; punchTimeAtTimestamp(finalValueWallClockTime, additionalTimestampForFinalPartition); enqueueWriteCommitted(TABLE_CONS, minTsForFinePartition(9_999_999)); punchTimeAtTimestamp(Long.MAX_VALUE, minTsForFinePartition(9_999_999)); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); long timestampEndOfBatchOne = minTsForFinePartition(permittedPartitions.get(readBatchSize - 1)) + LOW_TS; assertReadAtTimestampReturnsSentinel(TABLE_CONS, timestampEndOfBatchOne); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, timestampEndOfBatchOne); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(1); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); long timestampEndOfBatchTwo = minTsForFinePartition(permittedPartitions.get(2 * readBatchSize - 1)) + LOW_TS; assertReadAtTimestampReturnsSentinel(TABLE_CONS, timestampEndOfBatchTwo); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, timestampEndOfBatchTwo); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(2); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); long lastSweepableTimestamp = minTsForFinePartition(permittedPartitions.get(2 * readBatchSize)) + LOW_TS + 1; assertReadAtTimestampReturnsSentinel(TABLE_CONS, lastSweepableTimestamp); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, lastSweepableTimestamp); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(3); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(2 * readBatchSize + 2); assertThat(metricsManager).hasLastSweptTimestampConservativeEqualTo( maxTsForFinePartition(permittedPartitions.get(2 * readBatchSize))); assertThat(metricsManager).containsEntriesReadInBatchConservative(readBatchSize, readBatchSize, 2L); assertThat(metricsManager).hasEntriesReadInBatchMeanConservativeEqualTo((2 * readBatchSize + 2) / 3.0); setTimelockTime(5000L); assertThat(metricsManager).hasMillisSinceLastSweptConservativeEqualTo(5000L - finalValueWallClockTime); }
@Test public void sweepProgressesAcrossCoarsePartitions() { setSweepTimestamp(Long.MAX_VALUE); List<Integer> permittedPartitions = Lists.newArrayList(); permittedPartitions.add(0); enqueueWriteCommitted(TABLE_CONS, LOW_TS); for (int index = 0; index <= 3 * readBatchSize; index++) { int partitionToUse = 3 * index + 1; enqueueWriteCommitted(TABLE_CONS, minTsForCoarsePartition(partitionToUse) + LOW_TS); permittedPartitions.add(partitionToUse); } sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + minTsForCoarsePartition(permittedPartitions.get(readBatchSize - 1))); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS + minTsForCoarsePartition(permittedPartitions.get(readBatchSize))); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + minTsForCoarsePartition(permittedPartitions.get(2 * readBatchSize - 1))); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS + minTsForCoarsePartition(permittedPartitions.get(2 * readBatchSize))); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, LOW_TS + minTsForCoarsePartition(permittedPartitions.get(3 * readBatchSize - 1))); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, LOW_TS + minTsForCoarsePartition(permittedPartitions.get(3 * readBatchSize))); }
@Test public void sweepProgressesToJustBeforeSweepTsWhenNothingToSweep() { sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertProgressUpdatedToTimestamp(getSweepTsCons() - 1L); }
@Test public void sweepProgressesToEndOfPartitionWhenFewValuesAndSweepTsLarge() { long writeTs = getSweepTsCons() - 3 * TS_FINE_GRANULARITY; enqueueWriteCommitted(TABLE_CONS, writeTs); enqueueWriteCommitted(TABLE_CONS, writeTs + 5); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertProgressUpdatedToTimestamp(maxTsForFinePartition(tsPartitionFine(writeTs))); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertProgressUpdatedToTimestamp(getSweepTsCons() - 1L); }
@Test public void sweepCellOnlyOnceWhenInLastPartitionBeforeSweepTs() { immutableTs = 2 * TS_COARSE_GRANULARITY - TS_FINE_GRANULARITY; verify(spiedKvs, never()).deleteAllTimestamps(any(TableReference.class), anyMap()); enqueueWriteCommitted(TABLE_CONS, immutableTs - 1); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); verify(spiedKvs, times(1)).deleteAllTimestamps(any(TableReference.class), anyMap()); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); verify(spiedKvs, times(1)).deleteAllTimestamps(any(TableReference.class), anyMap()); }
@Test public void sweepableTimestampsGetsScrubbedWhenNoMoreToSweepButSweepTsInNewCoarsePartition() { long largestFirstPartitionCoarse = TS_COARSE_GRANULARITY - 1L; for (int i = 0; i < 2 * readBatchSize; i++) { enqueueWriteCommitted(TABLE_CONS, LOW_TS + TS_FINE_GRANULARITY * i); } enqueueWriteCommitted(TABLE_CONS, largestFirstPartitionCoarse); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertLowestFinePartitionInSweepableTimestampsEquals(tsPartitionFine(LOW_TS)); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertLowestFinePartitionInSweepableTimestampsEquals(tsPartitionFine(LOW_TS)); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertNoEntriesInSweepableTimestampsBeforeSweepTimestamp(); }
@Test public void sweepableTimestampsGetsScrubbedWhenLastSweptProgressesInNewCoarsePartition() { for (int i = 0; i < 2 * readBatchSize; i++) { enqueueWriteCommitted(TABLE_CONS, LOW_TS + TS_FINE_GRANULARITY * i); } enqueueWriteCommitted(TABLE_CONS, TS_COARSE_GRANULARITY - 1L); enqueueWriteCommitted(TABLE_CONS, 2 * TS_COARSE_GRANULARITY); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertLowestFinePartitionInSweepableTimestampsEquals(tsPartitionFine(LOW_TS)); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertLowestFinePartitionInSweepableTimestampsEquals(tsPartitionFine(LOW_TS)); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertLowestFinePartitionInSweepableTimestampsEquals(tsPartitionFine(2 * TS_COARSE_GRANULARITY)); }
@Test public void sweepableCellsGetsScrubbedWheneverPartitionIsCompletelySwept() { for (int i = 0; i < readBatchSize; i++) { long referenceTimestamp = LOW_TS + SweepQueueUtils.minTsForFinePartition(i); enqueueWriteCommitted(TABLE_CONS, referenceTimestamp); enqueueWriteCommitted(TABLE_CONS, referenceTimestamp + 1L); enqueueAtLeastThresholdWritesInDefaultShardWithStartTs(100, referenceTimestamp + 2L); putTimestampIntoTransactionTable(referenceTimestamp + 2L, referenceTimestamp + 2L); } for (int i = readBatchSize; i < 2 * readBatchSize; i++) { enqueueWriteCommitted(TABLE_CONS, SweepQueueUtils.minTsForFinePartition(i)); } enqueueWriteCommitted(TABLE_CONS, getSweepTsCons()); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertSweepableCellsHasNoEntriesInPartitionOfTimestamp(LOW_TS + 1); assertSweepableCellsHasEntryForTimestamp(SweepQueueUtils.minTsForFinePartition(readBatchSize)); assertSweepableCellsHasEntryForTimestamp(getSweepTsCons()); assertSweepableCellsHasNoDedicatedRowsForShard(CONS_SHARD); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertSweepableCellsHasNoEntriesInPartitionOfTimestamp(LOW_TS + 1); assertSweepableCellsHasNoEntriesInPartitionOfTimestamp(SweepQueueUtils.minTsForFinePartition(readBatchSize)); assertSweepableCellsHasEntryForTimestamp(getSweepTsCons()); assertSweepableCellsHasNoDedicatedRowsForShard(CONS_SHARD); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertSweepableCellsHasNoEntriesInPartitionOfTimestamp(LOW_TS + 1); assertSweepableCellsHasNoEntriesInPartitionOfTimestamp(SweepQueueUtils.minTsForFinePartition(readBatchSize)); assertSweepableCellsHasEntryForTimestamp(getSweepTsCons()); assertSweepableCellsHasNoDedicatedRowsForShard(CONS_SHARD); }
@Test public void doesNotTransitivelyRetainWritesFromBeforeSweepTimestamp() { long sweepTimestamp = getSweepTsCons(); enqueueWriteCommitted(TABLE_CONS, sweepTimestamp - 10); enqueueTombstone(TABLE_CONS, sweepTimestamp - 5); enqueueWriteCommitted(TABLE_CONS, sweepTimestamp + 5); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertReadAtTimestampReturnsSentinel(TABLE_CONS, sweepTimestamp - 5); assertReadAtTimestampReturnsTombstoneAtTimestamp(TABLE_CONS, sweepTimestamp - 5 + 1, sweepTimestamp - 5); assertTestValueEnqueuedAtGivenTimestampStillPresent(TABLE_CONS, sweepTimestamp + 5); assertThat(metricsManager).hasTombstonesPutConservativeEqualTo(1); assertThat(metricsManager).hasLastSweptTimestampConservativeEqualTo( sweepTimestamp - 1); }
@Test public void testSweepTimestampMetric() { unreadableTs = 17; immutableTs = 40; sweepNextBatch(ShardAndStrategy.conservative(0)); sweepNextBatch(ShardAndStrategy.thorough(0)); assertThat(metricsManager).hasSweepTimestampConservativeEqualTo(17L); assertThat(metricsManager).hasSweepTimestampThoroughEqualTo(40); immutableTs = 5; sweepNextBatch(ShardAndStrategy.conservative(0)); sweepNextBatch(ShardAndStrategy.thorough(0)); assertThat(metricsManager).hasSweepTimestampConservativeEqualTo(5L); assertThat(metricsManager).hasSweepTimestampThoroughEqualTo(5); }
@Test public void doNotSweepAnythingAfterEntryWithCommitTsAfterSweepTs() { immutableTs = 1000L; enqueueWriteCommitted(TABLE_CONS, 900); enqueueWriteCommitted(TABLE_CONS, 910); enqueueWriteCommitted(TABLE_CONS, 920); enqueueWriteCommitedAt(TABLE_CONS, 950, immutableTs); enqueueWriteCommitted(TABLE_CONS, 970); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(950 - 1); ArgumentCaptor<Map> argument = ArgumentCaptor.forClass(Map.class); verify(spiedKvs, times(1)).deleteAllTimestamps(eq(TABLE_CONS), argument.capture()); assertThat(argument.getValue()).containsValue( new TimestampRangeDelete.Builder().timestamp(920L).endInclusive(false).deleteSentinels(false).build()); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(950 - 1); verify(spiedKvs, times(1)).deleteAllTimestamps(any(TableReference.class), anyMap()); immutableTs = 1001L; sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(1001L - 1L); verify(spiedKvs, times(2)).deleteAllTimestamps(eq(TABLE_CONS), argument.capture()); assertThat(argument.getValue()).containsValue( new TimestampRangeDelete.Builder().timestamp(970L).endInclusive(false).deleteSentinels(false).build()); }
@Test public void doNotDeleteAnythingAfterEntryWithCommitTsAfterSweepTs() { immutableTs = 1000L; enqueueWriteUncommitted(TABLE_CONS, 900); enqueueWriteUncommitted(TABLE_CONS, 920); enqueueWriteCommitedAt(TABLE_CONS, 950, 2000); enqueueWriteUncommitted(TABLE_CONS, 970); enqueueWriteUncommitted(TABLE_CONS, 1110); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(950 - 1); verify(spiedKvs, never()).deleteAllTimestamps(any(TableReference.class), anyMap()); ArgumentCaptor<Multimap> multimap = ArgumentCaptor.forClass(Multimap.class); verify(spiedKvs, times(1)).delete(eq(TABLE_CONS), multimap.capture()); assertThat(multimap.getValue().keySet()).containsExactly(DEFAULT_CELL); assertThat(multimap.getValue().values()).containsExactly(900L, 920L); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(950 - 1); verify(spiedKvs, never()).deleteAllTimestamps(any(TableReference.class), anyMap()); verify(spiedKvs, times(1)).delete(any(TableReference.class), any(Multimap.class)); assertReadAtTimestampReturnsValue(TABLE_CONS, 1500L, 1110L); immutableTs = 2009L; sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(2009L - 1L); ArgumentCaptor<Map> map = ArgumentCaptor.forClass(Map.class); verify(spiedKvs, times(1)).deleteAllTimestamps(eq(TABLE_CONS), map.capture()); assertThat(map.getValue()).containsValue( new TimestampRangeDelete.Builder().timestamp(950L).endInclusive(false).deleteSentinels(false).build()); assertReadAtTimestampReturnsValue(TABLE_CONS, 1500L, 950L); }
@Test public void stopReadingEarlyWhenEncounteringEntryKnownToBeCommittedAfterSweepTs() { immutableTs = 100L; enqueueWriteCommitted(TABLE_CONS, 10); enqueueWriteCommitedAt(TABLE_CONS, 30, 150); putTimestampIntoTransactionTable(50, 200); Map<Integer, Integer> largeWriteDistribution = enqueueAtLeastThresholdWritesInDefaultShardWithStartTs(100, 50); int writesInDedicated = largeWriteDistribution.get(CONS_SHARD); enqueueWriteUncommitted(TABLE_CONS, 70); enqueueWriteCommitted(TABLE_CONS, 90); assertThat(sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(4 + writesInDedicated); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated + 1); immutableTs = 170; sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated + 3); immutableTs = 250; sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(4 + writesInDedicated + 3 + writesInDedicated + 2); }
@Test public void stopReadingEarlyInOtherShardWhenEncounteringEntryKnownToBeCommittedAfterSweepTs() { immutableTs = 100L; putTimestampIntoTransactionTable(50, 200); Map<Integer, Integer> largeWriteDistribution = enqueueAtLeastThresholdWritesInDefaultShardWithStartTs(100, 50); int writesInDedicated = largeWriteDistribution.get(CONS_SHARD); int otherShard = IntMath.mod(CONS_SHARD + 1, DEFAULT_SHARDS); int writesInOther = largeWriteDistribution.get(otherShard); assertThat(writesInOther).isGreaterThan(0); assertThat(sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD))).isEqualTo(writesInDedicated); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(writesInDedicated); assertThat(sweepNextBatch(ShardAndStrategy.conservative(otherShard))).isEqualTo(1); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(writesInDedicated + 1); immutableTs = 250; assertThat(sweepNextBatch(ShardAndStrategy.conservative(otherShard))).isEqualTo(writesInOther); assertThat(metricsManager).hasEntriesReadConservativeEqualTo(writesInDedicated + 1 + writesInOther); }
@Test public void batchIncludesAllWritesWithTheSameTimestampAndDoesNotSkipOrRepeatAnyWritesInNextIteration() { TargetedSweeper sweeperConservative = getSingleShardSweeper(); int relativePrime = MAX_CELLS_GENERIC - 1; assertThat(SWEEP_BATCH_SIZE % relativePrime).isNotEqualTo(0); int minTsToReachBatchSize = (SWEEP_BATCH_SIZE - 1) / relativePrime + 1; commitTransactionsWithWritesIntoUniqueCells(minTsToReachBatchSize + 1, relativePrime, sweeperConservative); sweepNextBatch(sweeperConservative, ShardAndStrategy.conservative(0)); ArgumentCaptor<Map> map = ArgumentCaptor.forClass(Map.class); verify(spiedKvs, atLeast(1)).deleteAllTimestamps(eq(TABLE_CONS), map.capture()); assertThat(map.getAllValues().stream().map(Map::size).mapToInt(x -> x).sum()) .isEqualTo(minTsToReachBatchSize * relativePrime); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(0))) .isEqualTo(1000 + minTsToReachBatchSize - 1); sweepNextBatch(sweeperConservative, ShardAndStrategy.conservative(0)); verify(spiedKvs, atLeast(2)).deleteAllTimestamps(eq(TABLE_CONS), map.capture()); assertThat(map.getValue().size()).isEqualTo(relativePrime); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(0))) .isEqualTo(maxTsForFinePartition(0)); }
@Test public void deletesGetBatched() { TargetedSweeper sweeperConservative = getSingleShardSweeper(); int numberOfTimestamps = 5 * BATCH_SIZE_KVS / MAX_CELLS_GENERIC + 1; commitTransactionsWithWritesIntoUniqueCells(numberOfTimestamps, MAX_CELLS_GENERIC, sweeperConservative); sweepNextBatch(sweeperConservative, ShardAndStrategy.conservative(0)); ArgumentCaptor<Map> map = ArgumentCaptor.forClass(Map.class); verify(spiedKvs, times(6)).deleteAllTimestamps(eq(TABLE_CONS), map.capture()); assertThat(map.getAllValues().stream().map(Map::size).mapToInt(x -> x).sum()) .isEqualTo(5 * BATCH_SIZE_KVS + MAX_CELLS_GENERIC); assertThat(progress.getLastSweptTimestamp(ShardAndStrategy.conservative(0))) .isEqualTo(maxTsForFinePartition(0)); }
@Test public void doesNotLeaveSentinelsIfTableDestroyed() { enqueueWriteCommitted(TABLE_CONS, 10); immutableTs = 11; unreadableTs = 11; spiedKvs.truncateTable(TABLE_CONS); assertThat(spiedKvs.getRange(TABLE_CONS, RangeRequest.all(), Long.MAX_VALUE)).isExhausted(); sweepNextBatch(ShardAndStrategy.conservative(CONS_SHARD)); assertThat(spiedKvs.getRange(TABLE_CONS, RangeRequest.all(), Long.MAX_VALUE)).isExhausted(); }