private Collection<SSTableReader> doAntiCompaction(ColumnFamilyStore cfs, Collection<Range<Token>> ranges, Collection<SSTableReader> repairedSSTables, long repairedAt)
    {
        List<SSTableReader> anticompactedSSTables = new ArrayList<>();
        int repairedKeyCount = 0;
        int unrepairedKeyCount = 0;
        logger.info("Performing anticompaction on {} sstables", repairedSSTables.size());
        // iterate over sstables to check if the repaired / unrepaired ranges intersect them.
        for (SSTableReader sstable : repairedSSTables)
        {
            // check that compaction hasn't stolen any sstables used in previous repair sessions
            // if we need to skip the anticompaction, it will be carried out by the next repair
            if (!new File(sstable.getFilename()).exists())
            {
                logger.info("Skipping anticompaction for {}, required sstable was compacted and is no longer available.", sstable);
                continue;
            }

            logger.info("Anticompacting {}", sstable);
            Set<SSTableReader> sstableAsSet = new HashSet<>();
            sstableAsSet.add(sstable);

            File destination = cfs.directories.getWriteableLocationAsFile(cfs.getExpectedCompactedFileSize(sstableAsSet, OperationType.ANTICOMPACTION));
            SSTableRewriter repairedSSTableWriter = new SSTableRewriter(cfs, sstableAsSet, sstable.maxDataAge, false);
            SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter(cfs, sstableAsSet, sstable.maxDataAge, false);

            try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(new HashSet<>(Collections.singleton(sstable)));
                 CompactionController controller = new CompactionController(cfs, sstableAsSet, CFMetaData.DEFAULT_GC_GRACE_SECONDS))
            {
                int expectedBloomFilterSize = Math.max(cfs.metadata.getMinIndexInterval(), (int)sstable.estimatedKeys());
                repairedSSTableWriter.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, repairedAt, sstable));
                unRepairedSSTableWriter.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, ActiveRepairService.UNREPAIRED_SSTABLE, sstable));

                CompactionIterable ci = new CompactionIterable(OperationType.ANTICOMPACTION, scanners.scanners, controller);
                Iterator<AbstractCompactedRow> iter = ci.iterator();
                metrics.beginCompaction(ci);
                try
                {
                    while (iter.hasNext())
                    {
                        AbstractCompactedRow row = iter.next();
                        // if current range from sstable is repaired, save it into the new repaired sstable
                        if (Range.isInRanges(row.key.getToken(), ranges))
                        {
                            repairedSSTableWriter.append(row);
                            repairedKeyCount++;
                        }
                        // otherwise save into the new 'non-repaired' table
                        else
                        {
                            unRepairedSSTableWriter.append(row);
                            unrepairedKeyCount++;
                        }
                    }
                }
                finally
                {
                    metrics.finishCompaction(ci);
                }
                anticompactedSSTables.addAll(repairedSSTableWriter.finish(repairedAt));
                anticompactedSSTables.addAll(unRepairedSSTableWriter.finish(ActiveRepairService.UNREPAIRED_SSTABLE));
                cfs.getDataTracker().markCompactedSSTablesReplaced(sstableAsSet, anticompactedSSTables, OperationType.ANTICOMPACTION);
            }
            catch (Throwable e)
            {
                JVMStabilityInspector.inspectThrowable(e);
                logger.error("Error anticompacting " + sstable, e);
                repairedSSTableWriter.abort();
                unRepairedSSTableWriter.abort();
            }
        }
        String format = "Repaired {} keys of {} for {}/{}";
        logger.debug(format, repairedKeyCount, (repairedKeyCount + unrepairedKeyCount), cfs.keyspace, cfs.getColumnFamilyName());
        String format2 = "Anticompaction completed successfully, anticompacted from {} to {} sstable(s).";
        logger.info(format2, repairedSSTables.size(), anticompactedSSTables.size());

        return anticompactedSSTables;
    }
private Collection < SSTableReader > doAntiCompaction ( ColumnFamilyStore cfs , Collection < Range < Token > > ranges , Collection < SSTableReader > repairedSSTables , long repairedAt ) { List < SSTableReader > anticompactedSSTables = new ArrayList < > ( ) ; int repairedKeyCount = 0 ; int unrepairedKeyCount = 0 ; logger . info ( "Performing anticompaction on {} sstables" , repairedSSTables . size ( ) ) ; for ( SSTableReader sstable : repairedSSTables ) { if ( ! new File ( sstable . getFilename ( ) ) . exists ( ) ) { logger . info ( "Skipping anticompaction for {}, required sstable was compacted and is no longer available." , sstable ) ; continue ; } logger . info ( "Anticompacting {}" , sstable ) ; Set < SSTableReader > sstableAsSet = new HashSet < > ( ) ; sstableAsSet . add ( sstable ) ; File destination = cfs . directories . getWriteableLocationAsFile ( cfs . getExpectedCompactedFileSize ( sstableAsSet , OperationType . ANTICOMPACTION ) ) ; SSTableRewriter repairedSSTableWriter = new SSTableRewriter ( cfs , sstableAsSet , sstable . maxDataAge , false ) ; SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter ( cfs , sstableAsSet , sstable . maxDataAge , false ) ; try ( AbstractCompactionStrategy . ScannerList scanners = cfs . getCompactionStrategy ( ) . getScanners ( new HashSet < > ( Collections . singleton ( sstable ) ) ) ; CompactionController controller = new CompactionController ( cfs , sstableAsSet , CFMetaData . DEFAULT_GC_GRACE_SECONDS ) ) { int expectedBloomFilterSize = Math . max ( cfs . metadata . getMinIndexInterval ( ) , ( int ) sstable . estimatedKeys ( ) ) ; repairedSSTableWriter . switchWriter ( CompactionManager . createWriter ( cfs , destination , expectedBloomFilterSize , repairedAt , sstable ) ) ; unRepairedSSTableWriter . switchWriter ( CompactionManager . createWriter ( cfs , destination , expectedBloomFilterSize , ActiveRepairService . UNREPAIRED_SSTABLE , sstable ) ) ; CompactionIterable ci = new CompactionIterable ( OperationType . ANTICOMPACTION , scanners . scanners , controller ) ; Iterator < AbstractCompactedRow > iter = ci . iterator ( ) ; metrics . beginCompaction ( ci ) ; try { while ( iter . hasNext ( ) ) { AbstractCompactedRow row = iter . next ( ) ; if ( Range . isInRanges ( row . key . getToken ( ) , ranges ) ) { repairedSSTableWriter . append ( row ) ; repairedKeyCount ++ ; } else { unRepairedSSTableWriter . append ( row ) ; unrepairedKeyCount ++ ; } } } finally { metrics . finishCompaction ( ci ) ; } anticompactedSSTables . addAll ( repairedSSTableWriter . finish ( repairedAt ) ) ; anticompactedSSTables . addAll ( unRepairedSSTableWriter . finish ( ActiveRepairService . UNREPAIRED_SSTABLE ) ) ; cfs . getDataTracker ( ) . markCompactedSSTablesReplaced ( sstableAsSet , anticompactedSSTables , OperationType . ANTICOMPACTION ) ; } catch ( Throwable e ) { JVMStabilityInspector . inspectThrowable ( e ) ; logger . error ( "Error anticompacting " + sstable , e ) ; repairedSSTableWriter . abort ( ) ; unRepairedSSTableWriter . abort ( ) ; } } String format = "Repaired {} keys of {} for {}/{}" ; logger . debug ( format , repairedKeyCount , ( repairedKeyCount + unrepairedKeyCount ) , cfs . keyspace , cfs . getColumnFamilyName ( ) ) ; String format2 = "Anticompaction completed successfully, anticompacted from {} to {} sstable(s)." ; logger . info ( format2 , repairedSSTables . size ( ) , anticompactedSSTables . size ( ) ) ; return anticompactedSSTables ; }
Splits up an sstable into two new sstables. The first of the new tables will store repaired ranges, the second
will store the non-repaired ranges. Once anticompation is completed, the original sstable is marked as compacted
and subsequently deleted.
@param cfs
@param repairedSSTables
@param ranges Repaired ranges to be placed into one of the new sstables. The repaired table will be tracked via
the {@link org.apache.cassandra.io.sstable.metadata.StatsMetadata#repairedAt} field.