import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * We make sure to close mergedIterator in close() and CompactionIterator is itself an AutoCloseable.
 *
 * @param scanners   - a list of sstable scanners
 * @param cfMetaData - CF Metadata for those sstables
 * @param nowInSec   - now in secs
 */
@SuppressWarnings("resource")
public SSTableIterator(final List<ISSTableScanner> scanners, final CFMetaData cfMetaData, final int nowInSec) {
    this.cfMetaData = cfMetaData;
    this.scanners = scanners;
    this.nowInSec = nowInSec;
    this.bytesRead = 0;

    long bytes = 0;
    for (ISSTableScanner scanner : scanners) {
        bytes += scanner.getLengthInBytes();
    }

    this.totalBytes = bytes;
    this.mergeCounters = new long[scanners.size()];

    final UnfilteredPartitionIterator merged = scanners.isEmpty()
            ? EmptyIterators.unfilteredPartition(cfMetaData, false)
            : UnfilteredPartitionIterators.merge(scanners, nowInSec, listener());

    UnfilteredPartitionIterator mergeAndPurge = Transformation.apply(merged, new Purger(nowInSec));
    compacted = UnfilteredPartitionIterators.filter(mergeAndPurge, nowInSec);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
@Test
public void testConvertingSSTable() throws IOException {
    final String inputFile = CASS3_DATA_DIR + "keyspace1/bills_compress/mc-6-big-Data.db";
    final CFMetaData inputCFMetaData = SSTableUtils.metaDataFromSSTable(inputFile,
                                                                        TestBaseSSTableFunSuite.HADOOP_CONF);
    final CassandraTable cassandraTable = new CassandraTable.CassandraTableBuilder()
                                                            .withClusterName("cass_share")
                                                            .withKeyspaceName("casspactor")
                                                            .withTableName("bills_compress")
                                                            .build();
    final String outputLocation = "/tmp";

    final SSTableSingleWriter<UnfilteredRowIterator> writer =
            new SSTableSingleWriter(inputCFMetaData, cassandraTable, outputLocation,
                                    TestBaseSSTableFunSuite.HADOOP_CONF);

    SSTableSingleReader reader = new SSTableSingleReader(inputFile, cassandraTable,
                                                         TestBaseSSTableFunSuite.HADOOP_CONF);
    final ISSTableScanner currentScanner = reader.getSSTableScanner();

    writer.write(currentScanner);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * Test on the SSTableIterator.
 * @throws IOException
 */
@Test
public void testCasspactorIterator() throws IOException {
    final String inputSSTableFullPathFileName = CASS3_DATA_DIR + "keyspace1/bills_compress/mc-6-big-Data.db";
    final SSTableSingleReader reader1 = new SSTableSingleReader(inputSSTableFullPathFileName,
                                                               TestBaseSSTableFunSuite.HADOOP_CONF);
    final SSTableSingleReader reader2 = new SSTableSingleReader(inputSSTableFullPathFileName,
                                                 TestBaseSSTableFunSuite.HADOOP_CONF);
    final CFMetaData cfMetaData = reader1.getCfMetaData();
    final List<ISSTableScanner> scanners = new ArrayList<>();
    final int nowInSecs = (int) (System.currentTimeMillis() / 1000);

    scanners.add(reader1.getSSTableScanner());
    scanners.add(reader2.getSSTableScanner());

    int counter = 0;
    try (SSTableIterator ci = new SSTableIterator(scanners, reader1.getCfMetaData(), nowInSecs)) {
        while (ci.hasNext()) {
            final RowIterator rowIterator = ci.next();
            counter += printRowDetails(cfMetaData, rowIterator, false);
        }
    }

    Assert.assertEquals(4, counter);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
private int getRowCount(final String inputSSTableFullPathFileName, String cql,
                        boolean isThriftTable) throws IOException {
    LOGGER.info("Input file name: " + inputSSTableFullPathFileName);
    int counter = 0;
    CFMetaData cfMetaData = CFMetaData.compile(cql, "casspactor2");
    final SSTableSingleReader sstableSingleReader =
            new SSTableSingleReader(inputSSTableFullPathFileName, cfMetaData, TestBaseSSTableFunSuite.HADOOP_CONF);
    final ISSTableScanner currentScanner =
            sstableSingleReader.getSSTableScanner(Long.MIN_VALUE, Long.MAX_VALUE);

    final int nowInSecs = (int) (System.currentTimeMillis() / 1000);
    final List<ISSTableScanner> scanners = new ArrayList<>();
    scanners.add(currentScanner);
    try (SSTableIterator ci = new SSTableIterator(scanners, cfMetaData, nowInSecs)) {
        while (ci.hasNext()) {
            final RowIterator rowIterator = ci.next();
            counter += printRowDetails(cfMetaData, rowIterator, isThriftTable);
        }
    }

    return counter;
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * Returns a list of KeyScanners given sstables and a range on which to scan.
 * The default implementation simply grab one SSTableScanner per-sstable, but overriding this method
 * allow for a more memory efficient solution if we know the sstable don't overlap (see
 * LeveledCompactionStrategy for instance).
 */
public ScannerList getScanners(Collection<SSTableReader> sstables, Range<Token> range)
{
    RateLimiter limiter = CompactionManager.instance.getRateLimiter();
    ArrayList<ISSTableScanner> scanners = new ArrayList<ISSTableScanner>();
    try
    {
        for (SSTableReader sstable : sstables)
            scanners.add(sstable.getScanner(range, limiter));
    }
    catch (Throwable t)
    {
        try
        {
            new ScannerList(scanners).close();
        }
        catch (Throwable t2)
        {
            t.addSuppressed(t2);
        }
        throw t;
    }
    return new ScannerList(scanners);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
public void close()
{
    Throwable t = null;
    for (ISSTableScanner scanner : scanners)
    {
        try
        {
            scanner.close();
        }
        catch (Throwable t2)
        {
            JVMStabilityInspector.inspectThrowable(t2);
            if (t == null)
                t = t2;
            else
                t.addSuppressed(t2);
        }
    }
    if (t != null)
        throw Throwables.propagate(t);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
@Override
public synchronized ScannerList getScanners(Collection<SSTableReader> sstables, Range<Token> range)
{
    List<SSTableReader> repairedSSTables = new ArrayList<>();
    List<SSTableReader> unrepairedSSTables = new ArrayList<>();
    for (SSTableReader sstable : sstables)
        if (sstable.isRepaired())
            repairedSSTables.add(sstable);
        else
            unrepairedSSTables.add(sstable);
    ScannerList repairedScanners = repaired.getScanners(repairedSSTables, range);
    ScannerList unrepairedScanners = unrepaired.getScanners(unrepairedSSTables, range);
    List<ISSTableScanner> scanners = new ArrayList<>(repairedScanners.scanners.size() + unrepairedScanners.scanners.size());
    scanners.addAll(repairedScanners.scanners);
    scanners.addAll(unrepairedScanners.scanners);
    return new ScannerList(scanners);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * Create ISSTableScanner from the given sstables
 *
 * Delegates the call to the compaction strategies to allow LCS to create a scanner
 * @param sstables
 * @param ranges
 * @return
 */
@SuppressWarnings("resource")
public synchronized AbstractCompactionStrategy.ScannerList getScanners(Collection<SSTableReader> sstables,  Collection<Range<Token>> ranges)
{
    List<SSTableReader> repairedSSTables = new ArrayList<>();
    List<SSTableReader> unrepairedSSTables = new ArrayList<>();
    for (SSTableReader sstable : sstables)
    {
        if (sstable.isRepaired())
            repairedSSTables.add(sstable);
        else
            unrepairedSSTables.add(sstable);
    }

    Set<ISSTableScanner> scanners = new HashSet<>(sstables.size());
    AbstractCompactionStrategy.ScannerList repairedScanners = repaired.getScanners(repairedSSTables, ranges);
    AbstractCompactionStrategy.ScannerList unrepairedScanners = unrepaired.getScanners(unrepairedSSTables, ranges);
    scanners.addAll(repairedScanners.scanners);
    scanners.addAll(unrepairedScanners.scanners);
    return new AbstractCompactionStrategy.ScannerList(new ArrayList<>(scanners));
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
@SuppressWarnings("resource") // We make sure to close mergedIterator in close() and CompactionIterator is itself an AutoCloseable
public CompactionIterator(OperationType type, List<ISSTableScanner> scanners, CompactionController controller, int nowInSec, UUID compactionId, CompactionMetrics metrics)
{
    this.controller = controller;
    this.type = type;
    this.scanners = scanners;
    this.nowInSec = nowInSec;
    this.compactionId = compactionId;
    this.bytesRead = 0;

    long bytes = 0;
    for (ISSTableScanner scanner : scanners)
        bytes += scanner.getLengthInBytes();
    this.totalBytes = bytes;
    this.mergeCounters = new long[scanners.size()];
    this.metrics = metrics;

    if (metrics != null)
        metrics.beginCompaction(this);

    UnfilteredPartitionIterator merged = scanners.isEmpty()
                                         ? EmptyIterators.unfilteredPartition(controller.cfs.metadata, false)
                                         : UnfilteredPartitionIterators.merge(scanners, nowInSec, listener());
    boolean isForThrift = merged.isForThrift(); // to stop capture of iterator in Purger, which is confusing for debug
    this.compacted = Transformation.apply(merged, new Purger(isForThrift, controller, nowInSec));
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * Returns a list of KeyScanners given sstables and a range on which to scan.
 * The default implementation simply grab one SSTableScanner per-sstable, but overriding this method
 * allow for a more memory efficient solution if we know the sstable don't overlap (see
 * LeveledCompactionStrategy for instance).
 */
@SuppressWarnings("resource")
public ScannerList getScanners(Collection<SSTableReader> sstables, Collection<Range<Token>> ranges)
{
    RateLimiter limiter = CompactionManager.instance.getRateLimiter();
    ArrayList<ISSTableScanner> scanners = new ArrayList<ISSTableScanner>();
    try
    {
        for (SSTableReader sstable : sstables)
            scanners.add(sstable.getScanner(ranges, limiter));
    }
    catch (Throwable t)
    {
        try
        {
            new ScannerList(scanners).close();
        }
        catch (Throwable t2)
        {
            t.addSuppressed(t2);
        }
        throw t;
    }
    return new ScannerList(scanners);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
public static ISSTableScanner getScanner(SSTableReader sstable, Collection<Range<Token>> tokenRanges, RateLimiter limiter)
{
    // We want to avoid allocating a SSTableScanner if the range don't overlap the sstable (#5249)
    List<Pair<Long, Long>> positions = sstable.getPositionsForRanges(tokenRanges);
    if (positions.isEmpty())
        return new EmptySSTableScanner(sstable);

    return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata), null, limiter, false, makeBounds(sstable, tokenRanges).iterator());
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * Direct I/O SSTableScanner over a defined range of tokens.
 *
 * @param range the range of keys to cover
 * @return A Scanner for seeking over the rows of the SSTable.
 */
public ISSTableScanner getScanner(Range<Token> range, RateLimiter limiter)
{
    if (range == null)
        return getScanner(limiter);
    return getScanner(Collections.singletonList(range), limiter);
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
private void updateBytesRead() {
    long n = 0;
    for (ISSTableScanner scanner : scanners) {
        n += scanner.getCurrentPosition();
    }
    bytesRead = n;
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * close the underneath iterator.
 */
public void close() {
    try {
        compacted.close();
        for (ISSTableScanner scanner : scanners) {
            scanner.close();
        }
    } catch (Exception e) {
        LOGGER.info(e.getMessage());
    }
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
public static void main(String[] args) {
    String inputSSTableFullPathFileName = new File(getInputFile(args)).getAbsolutePath();
    LOGGER.info("Input file name: " + inputSSTableFullPathFileName);
    Configuration conf = new Configuration();
    final Descriptor inputSSTableDescriptor = Descriptor.fromFilename(inputSSTableFullPathFileName, conf);
    SSTableWriter writer = null;

    try {
        final CFMetaData inputCFMetaData =
                SSTableUtils.metaDataFromSSTable(inputSSTableFullPathFileName, conf);
        final CFMetaData outputCFMetaData = SSTableUtils.createNewCFMetaData(inputSSTableDescriptor, inputCFMetaData);

        final SSTableReader inputSStable = SSTableReader.openNoValidation(inputSSTableDescriptor, inputCFMetaData);
        writer = SSTableUtils.createSSTableWriter(inputSSTableDescriptor, outputCFMetaData, inputSStable);

        final ISSTableScanner currentScanner = inputSStable.getScanner();

        while (currentScanner.hasNext()) {
            final UnfilteredRowIterator row = currentScanner.next();
            writer.append(row);
        }
        writer.finish(false);
    } catch (IOException e) {
        e.printStackTrace(System.err);
    } finally {
        FileUtils.closeQuietly(writer);
    }

}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/******************************************************
 * 1. Input data
 * This is the schema definition of the table that is used to generate the non-compressed input data:
 * <p>
 * CREATE TABLE bills_nc (
 * user text,
 * balance int static,
 * expense_id int,
 * amount int,
 * name text,
 * PRIMARY KEY (user, expense_id))
 * WITH compression = { 'sstable_compression' : '' };
 * <p>
 * <p>
 * 2. Compressing and producing output data
 * Running this main will convert data file under src/test/resources/data/bills_compress/mc-6-big-Data.db
 * in to the corresponding compressed file, using LZ4 compression, along with auxiliary
 * files (CompressionInfo.db, Index.db, etc).
 * <p>
 * The output is under cassanrda/compresseddata/cassandra/data directory
 * <p>
 * 3. Verification
 * Since this is C* 3.0 format, you should use sstabledump command to dump out the json contents
 * for both intput data and output data to verify.
 * %>sstabledump cassandra/data/mc-1-big-Data.db
 * and
 * %>sstabledump cassandra/compresseddata/cassandra/data/mc-1-big-Data.db
 *******************************************************/
@Test
public void testWritingToLocalSSTable() {
    final String inputSSTableFullPathFileName = CASS3_DATA_DIR + "keyspace1/bills_compress/mc-6-big-Data.db";
    LOGGER.info("Input file name: " + inputSSTableFullPathFileName);

    final Descriptor inputSSTableDescriptor = Descriptor.fromFilename(inputSSTableFullPathFileName,
                                                            TestBaseSSTableFunSuite.HADOOP_CONF);
    SSTableWriter writer = null;

    try {
        SSTableSingleReader reader = new SSTableSingleReader(inputSSTableFullPathFileName,
                                                             "casspactor",
                                                             "bills_nc",
                                                             TestBaseSSTableFunSuite.HADOOP_CONF);
        final CFMetaData inputCFMetaData = reader.getCfMetaData();
        final ISSTableScanner currentScanner = reader.getSSTableScanner();
        final SSTableReader inputSStable = reader.getSstableReader();

        //Create writer
        final CFMetaData outputCFMetaData = SSTableUtils.createNewCFMetaData(inputSSTableDescriptor, inputCFMetaData);
        writer = SSTableUtils.createSSTableWriter(inputSSTableDescriptor, outputCFMetaData, inputSStable);

        while (currentScanner.hasNext()) {
            final UnfilteredRowIterator row = currentScanner.next();
            writer.append(row);
        }
        writer.finish(false);
    } catch (IOException e) {
        e.printStackTrace(System.err);
    } finally {
        FileUtils.closeQuietly(writer);
    }
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
protected int getRowCount(final String inputSSTableFullPathFileName) {
    LOGGER.info("Input file name: " + inputSSTableFullPathFileName);
    int counter = 0;

    try {
        final SSTableSingleReader sstableSingleReader =
                new SSTableSingleReader(inputSSTableFullPathFileName, TestBaseSSTableFunSuite.HADOOP_CONF);
        final ISSTableScanner currentScanner =
                sstableSingleReader.getSSTableScanner(Long.MIN_VALUE, Long.MAX_VALUE);

        final CFMetaData cfMetaData = sstableSingleReader.getCfMetaData();
        final int nowInSecs = (int) (System.currentTimeMillis() / 1000);
        final List<ISSTableScanner> scanners = new ArrayList<>();
        scanners.add(currentScanner);
        try (SSTableIterator ci = new SSTableIterator(scanners, cfMetaData, nowInSecs)) {
            while (ci.hasNext()) {
                final RowIterator rowIterator = ci.next();
                counter += printRowDetails(cfMetaData, rowIterator, false);
            }
        }
    } catch (IOException e) {
        e.printStackTrace(System.err);
        return -1;
    }

    return counter;
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
/**
 * Construct a reader for Index.db sstable file.
 *
 * @param tableStats  SSTable statistics.
 */
public DataReader(SSTableStatistics tableStats, ISSTableScanner scanner, int gcGrace) {
    this.tableStats = tableStats;
    this.scanner = scanner;
    this.gcGrace = gcGrace;
    this.position = 0;
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
public AbstractCompactionIterable(CompactionController controller, OperationType type, List<ISSTableScanner> scanners)
{
    this.controller = controller;
    this.type = type;
    this.scanners = scanners;
    this.bytesRead = 0;

    long bytes = 0;
    for (ISSTableScanner scanner : scanners)
        bytes += scanner.getLengthInBytes();
    this.totalBytes = bytes;
    mergeCounters = new AtomicLong[scanners.size()];
    for (int i = 0; i < mergeCounters.length; i++)
        mergeCounters[i] = new AtomicLong();
}
 

import org.apache.cassandra.io.sstable.ISSTableScanner; //导入依赖的package包/类
private void updateBytesRead()
{
    long n = 0;
    for (ISSTableScanner scanner : scanners)
        n += scanner.getCurrentPosition();
    bytesRead = n;
}