@Test public void cleanupNonDedicatedRow() { useSingleShard(); writeCommittedConservativeRowForTimestamp(TS + 1, MAX_CELLS_GENERIC); writeCommittedConservativeRowForTimestamp(TS + 3, MAX_CELLS_GENERIC); writeCommittedConservativeRowForTimestamp(TS + 5, MAX_CELLS_GENERIC); sweepableCells.deleteNonDedicatedRows(ShardAndStrategy.conservative(0), ImmutableList.of(TS_FINE_PARTITION)); SweepableCellsRow row = SweepableCellsRow.of(TS_FINE_PARTITION, ImmutableTargetedSweepMetadata.builder() .conservative(true) .dedicatedRow(false) .shard(0) .dedicatedRowNumber(0) .build() .persistToBytes()); verifyRowsDeletedFromSweepQueue(ImmutableList.of(row)); }
void deleteNonDedicatedRows(ShardAndStrategy shardAndStrategy, Iterable<Long> partitionsFine) { List<byte[]> rows = Streams.stream(partitionsFine) .map(partitionFine -> computeRow(partitionFine, shardAndStrategy)) .map(SweepableCellsRow::persistToBytes) .collect(Collectors.toList()); deleteRows(rows); }
SweepableCells extends SweepQueueTable { void deleteNonDedicatedRows(ShardAndStrategy shardAndStrategy, Iterable<Long> partitionsFine) { List<byte[]> rows = Streams.stream(partitionsFine) .map(partitionFine -> computeRow(partitionFine, shardAndStrategy)) .map(SweepableCellsRow::persistToBytes) .collect(Collectors.toList()); deleteRows(rows); } }
SweepableCells extends SweepQueueTable { void deleteNonDedicatedRows(ShardAndStrategy shardAndStrategy, Iterable<Long> partitionsFine) { List<byte[]> rows = Streams.stream(partitionsFine) .map(partitionFine -> computeRow(partitionFine, shardAndStrategy)) .map(SweepableCellsRow::persistToBytes) .collect(Collectors.toList()); deleteRows(rows); } SweepableCells(
            KeyValueService kvs,
            WriteInfoPartitioner partitioner,
            TargetedSweepMetrics metrics,
            TransactionService transactionService); }
SweepableCells extends SweepQueueTable { void deleteNonDedicatedRows(ShardAndStrategy shardAndStrategy, Iterable<Long> partitionsFine) { List<byte[]> rows = Streams.stream(partitionsFine) .map(partitionFine -> computeRow(partitionFine, shardAndStrategy)) .map(SweepableCellsRow::persistToBytes) .collect(Collectors.toList()); deleteRows(rows); } SweepableCells(
            KeyValueService kvs,
            WriteInfoPartitioner partitioner,
            TargetedSweepMetrics metrics,
            TransactionService transactionService);  }
SweepableCells extends SweepQueueTable { void deleteNonDedicatedRows(ShardAndStrategy shardAndStrategy, Iterable<Long> partitionsFine) { List<byte[]> rows = Streams.stream(partitionsFine) .map(partitionFine -> computeRow(partitionFine, shardAndStrategy)) .map(SweepableCellsRow::persistToBytes) .collect(Collectors.toList()); deleteRows(rows); } SweepableCells(
            KeyValueService kvs,
            WriteInfoPartitioner partitioner,
            TargetedSweepMetrics metrics,
            TransactionService transactionService);   }