import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
protected List<Row> queryNextPage(int pageSize, ConsistencyLevel consistencyLevel, boolean localQuery)
throws RequestExecutionException
{
    SliceQueryFilter sf = (SliceQueryFilter)columnFilter;
    AbstractBounds<RowPosition> keyRange = lastReturnedKey == null ? command.keyRange : makeIncludingKeyBounds(lastReturnedKey);
    Composite start = lastReturnedName == null ? sf.start() : lastReturnedName;
    PagedRangeCommand pageCmd = new PagedRangeCommand(command.keyspace,
                                                      command.columnFamily,
                                                      command.timestamp,
                                                      keyRange,
                                                      sf,
                                                      start,
                                                      sf.finish(),
                                                      command.rowFilter,
                                                      pageSize,
                                                      command.countCQL3Rows);

    return localQuery
         ? pageCmd.executeLocally()
         : StorageProxy.getRangeSlice(pageCmd, consistencyLevel);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
/**
 * Convenience method that count (live) cells/rows for a given slice of a row, but page underneath.
 */
public static int countPaged(String keyspace,
                             String columnFamily,
                             ByteBuffer key,
                             SliceQueryFilter filter,
                             ConsistencyLevel consistencyLevel,
                             ClientState cState,
                             final int pageSize,
                             long now) throws RequestValidationException, RequestExecutionException
{
    SliceFromReadCommand command = new SliceFromReadCommand(keyspace, key, columnFamily, now, filter);
    final SliceQueryPager pager = new SliceQueryPager(command, consistencyLevel, cState, false);

    ColumnCounter counter = filter.columnCounter(Schema.instance.getCFMetaData(keyspace, columnFamily).comparator, now);
    while (!pager.isExhausted())
    {
        List<Row> next = pager.fetchPage(pageSize);
        if (!next.isEmpty())
            counter.countAll(next.get(0).cf);
    }
    return counter.live();
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
protected List<Row> queryNextPage(int pageSize, ConsistencyLevel consistencyLevel, boolean localQuery)
throws RequestValidationException, RequestExecutionException
{
    // For some queries, such as a DISTINCT query on static columns, the limit for slice queries will be lower
    // than the page size (in the static example, it will be 1).  We use the min here to ensure we don't fetch
    // more rows than we're supposed to.  See CASSANDRA-8108 for more details.
    SliceQueryFilter filter = command.filter.withUpdatedCount(Math.min(command.filter.count, pageSize));
    if (lastReturned != null)
        filter = filter.withUpdatedStart(lastReturned, cfm.comparator);

    logger.debug("Querying next page of slice query; new filter: {}", filter);
    ReadCommand pageCmd = command.withUpdatedFilter(filter);
    return localQuery
         ? Collections.singletonList(pageCmd.getRow(Keyspace.open(command.ksName)))
         : StorageProxy.read(Collections.singletonList(pageCmd), consistencyLevel, cstate);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
@Override
public ReadCommand maybeGenerateRetryCommand(RowDataResolver resolver, Row row)
{
    int maxLiveColumns = resolver.getMaxLiveCount();

    int count = filter.count;
    // We generate a retry if at least one node reply with count live columns but after merge we have less
    // than the total number of column we are interested in (which may be < count on a retry).
    // So in particular, if no host returned count live columns, we know it's not a short read.
    if (maxLiveColumns < count)
        return null;

    int liveCountInRow = row == null || row.cf == null ? 0 : filter.getLiveCount(row.cf, timestamp);
    if (liveCountInRow < getOriginalRequestedCount())
    {
        // We asked t (= count) live columns and got l (=liveCountInRow) ones.
        // From that, we can estimate that on this row, for x requested
        // columns, only l/t end up live after reconciliation. So for next
        // round we want to ask x column so that x * (l/t) == t, i.e. x = t^2/l.
        int retryCount = liveCountInRow == 0 ? count + 1 : ((count * count) / liveCountInRow) + 1;
        SliceQueryFilter newFilter = filter.withUpdatedCount(retryCount);
        return new RetriedSliceFromReadCommand(ksName, key, cfName, timestamp, newFilter, getOriginalRequestedCount());
    }

    return null;
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
private Composite makePrefix(CompositesIndex index, ByteBuffer key, ExtendedFilter filter, boolean isStart)
{
    if (key.remaining() == 0)
        return Composites.EMPTY;

    Composite prefix;
    IDiskAtomFilter columnFilter = filter.columnFilter(key);
    if (columnFilter instanceof SliceQueryFilter)
    {
        SliceQueryFilter sqf = (SliceQueryFilter)columnFilter;
        Composite columnName = isStart ? sqf.start() : sqf.finish();
        prefix = columnName.isEmpty() ? index.getIndexComparator().make(key) : index.makeIndexColumnPrefix(key, columnName);
    }
    else
    {
        prefix = index.getIndexComparator().make(key);
    }
    return isStart ? prefix.start() : prefix.end();
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
protected List<Row> queryNextPage(int pageSize, ConsistencyLevel consistencyLevel, boolean localQuery)
throws RequestExecutionException
{
    SliceQueryFilter sf = (SliceQueryFilter)columnFilter;
    AbstractBounds<RowPosition> keyRange = lastReturnedKey == null ? command.keyRange : makeIncludingKeyBounds(lastReturnedKey);
    ByteBuffer start = lastReturnedName == null ? sf.start() : lastReturnedName;
    PagedRangeCommand pageCmd = new PagedRangeCommand(command.keyspace,
                                                      command.columnFamily,
                                                      command.timestamp,
                                                      keyRange,
                                                      sf,
                                                      start,
                                                      sf.finish(),
                                                      command.rowFilter,
                                                      pageSize);

    return localQuery
         ? pageCmd.executeLocally()
         : StorageProxy.getRangeSlice(pageCmd, consistencyLevel);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
/**
 * Convenience method that count (live) cells/rows for a given slice of a row, but page underneath.
 */
public static int countPaged(String keyspace,
                             String columnFamily,
                             ByteBuffer key,
                             SliceQueryFilter filter,
                             ConsistencyLevel consistencyLevel,
                             final int pageSize,
                             long now) throws RequestValidationException, RequestExecutionException
{
    SliceFromReadCommand command = new SliceFromReadCommand(keyspace, key, columnFamily, now, filter);
    final SliceQueryPager pager = new SliceQueryPager(command, consistencyLevel, false);

    ColumnCounter counter = filter.columnCounter(Schema.instance.getComparator(keyspace, columnFamily), now);
    while (!pager.isExhausted())
    {
        List<Row> next = pager.fetchPage(pageSize);
        if (!next.isEmpty())
            counter.countAll(next.get(0).cf);
    }
    return counter.live();
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
@Override
public boolean intersects(List<ByteBuffer> minColumnNames, List<ByteBuffer> maxColumnNames, SliceQueryFilter filter)
{
    assert minColumnNames.size() == maxColumnNames.size();
    for (ColumnSlice slice : filter.slices)
    {
        ByteBuffer[] start = split(filter.isReversed() ? slice.finish : slice.start);
        ByteBuffer[] finish = split(filter.isReversed() ? slice.start : slice.finish);
        for (int i = 0; i < minColumnNames.size(); i++)
        {
            AbstractType<?> t = types.get(i);
            ByteBuffer s = i < start.length ? start[i] : ByteBufferUtil.EMPTY_BYTE_BUFFER;
            ByteBuffer f = i < finish.length ? finish[i] : ByteBufferUtil.EMPTY_BYTE_BUFFER;
            if (!t.intersects(minColumnNames.get(i), maxColumnNames.get(i), s, f))
                return false;
        }
    }
    return true;
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
/**
 * Convenience method that count (live) cells/rows for a given slice of a row, but page underneath.
 */
public static int countPaged(String keyspace,
                             String columnFamily,
                             ByteBuffer key,
                             SliceQueryFilter filter,
                             ConsistencyLevel consistencyLevel,
                             final int pageSize,
                             long now) throws RequestValidationException, RequestExecutionException
{
    SliceFromReadCommand command = new SliceFromReadCommand(keyspace, key, columnFamily, now, filter);
    final SliceQueryPager pager = new SliceQueryPager(command, consistencyLevel, false);

    ColumnCounter counter = filter.columnCounter(Schema.instance.getCFMetaData(keyspace, columnFamily).comparator, now);
    while (!pager.isExhausted())
    {
        List<Row> next = pager.fetchPage(pageSize);
        if (!next.isEmpty())
            counter.countAll(next.get(0).cf);
    }
    return counter.live();
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
@Override
public ReadCommand maybeGenerateRetryCommand(RowDataResolver resolver, Row row)
{
    int maxLiveColumns = resolver.getMaxLiveCount();

    int count = filter.count;
    // We generate a retry if at least one node reply with count live columns but after merge we have less
    // than the total number of column we are interested in (which may be < count on a retry).
    // So in particular, if no host returned count live columns, we know it's not a short read.
    if (maxLiveColumns < count)
        return null;

    int liveCountInRow = row == null || row.cf == null ? 0 : filter.getLiveCount(row.cf);
    if (liveCountInRow < getOriginalRequestedCount())
    {
        // We asked t (= count) live columns and got l (=liveCountInRow) ones.
        // From that, we can estimate that on this row, for x requested
        // columns, only l/t end up live after reconciliation. So for next
        // round we want to ask x column so that x * (l/t) == t, i.e. x = t^2/l.
        int retryCount = liveCountInRow == 0 ? count + 1 : ((count * count) / liveCountInRow) + 1;
        SliceQueryFilter newFilter = filter.withUpdatedCount(retryCount);
        return new RetriedSliceFromReadCommand(table, key, queryPath, newFilter, getOriginalRequestedCount());
    }

    return null;
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
protected List<Row> queryNextPage(int pageSize, ConsistencyLevel consistencyLevel, boolean localQuery)
throws RequestExecutionException
{
    SliceQueryFilter sf = (SliceQueryFilter)columnFilter;
    AbstractBounds<RowPosition> keyRange = lastReturnedKey == null ? command.keyRange : makeIncludingKeyBounds(lastReturnedKey);
    Composite start = lastReturnedName == null ? sf.start() : lastReturnedName;
    PagedRangeCommand pageCmd = new PagedRangeCommand(command.keyspace,
                                                      command.columnFamily,
                                                      command.timestamp,
                                                      keyRange,
                                                      sf,
                                                      start,
                                                      sf.finish(),
                                                      command.rowFilter,
                                                      pageSize);

    return localQuery
         ? pageCmd.executeLocally()
         : StorageProxy.getRangeSlice(pageCmd, consistencyLevel);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
private Composite makePrefix(CompositesIndex index, ByteBuffer key, ExtendedFilter filter, boolean isStart)
{
    if (key.remaining() == 0)
        return Composites.EMPTY;

    Composite prefix;
    IDiskAtomFilter columnFilter = filter.columnFilter(key);
    if (columnFilter instanceof SliceQueryFilter)
    {
        SliceQueryFilter sqf = (SliceQueryFilter)columnFilter;
        prefix = index.makeIndexColumnPrefix(key, isStart ? sqf.start() : sqf.finish());
    }
    else
    {
        prefix = index.getIndexComparator().make(key);
    }
    return isStart ? prefix.start() : prefix.end();
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
/**
 * Allows generic range paging with the slice column filter.
 * Typically, suppose we have rows A, B, C ... Z having each some columns in [1, 100].
 * And suppose we want to page through the query that for all rows returns the columns
 * within [25, 75]. For that, we need to be able to do a range slice starting at (row r, column c)
 * and ending at (row Z, column 75), *but* that only return columns in [25, 75].
 * That is what this method allows. The columnRange is the "window" of  columns we are interested
 * in each row, and columnStart (resp. columnEnd) is the start (resp. end) for the first
 * (resp. last) requested row.
 */
public ExtendedFilter makeExtendedFilter(AbstractBounds<RowPosition> keyRange,
                                         SliceQueryFilter columnRange,
                                         Composite columnStart,
                                         Composite columnStop,
                                         List<IndexExpression> rowFilter,
                                         int maxResults,
                                         boolean countCQL3Rows,
                                         long now)
{
    DataRange dataRange = new DataRange.Paging(keyRange, columnRange, columnStart, columnStop, metadata.comparator);
    return ExtendedFilter.create(this, dataRange, rowFilter, maxResults, countCQL3Rows, now);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
protected AbstractCType(boolean isByteOrderComparable)
{
    reverseComparator = new Comparator<Composite>()
    {
        public int compare(Composite c1, Composite c2)
        {
            return AbstractCType.this.compare(c2, c1);
        }
    };
    indexComparator = new Comparator<IndexInfo>()
    {
        public int compare(IndexInfo o1, IndexInfo o2)
        {
            return AbstractCType.this.compare(o1.lastName, o2.lastName);
        }
    };
    indexReverseComparator = new Comparator<IndexInfo>()
    {
        public int compare(IndexInfo o1, IndexInfo o2)
        {
            return AbstractCType.this.compare(o1.firstName, o2.firstName);
        }
    };

    serializer = new Serializer(this);

    indexSerializer = new IndexInfo.Serializer(this);
    sliceSerializer = new ColumnSlice.Serializer(this);
    sliceQueryFilterSerializer = new SliceQueryFilter.Serializer(this);
    deletionInfoSerializer = new DeletionInfo.Serializer(this);
    rangeTombstoneSerializer = new RangeTombstone.Serializer(this);
    rowIndexEntrySerializer = new RowIndexEntry.Serializer(this);
    this.isByteOrderComparable = isByteOrderComparable;
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
public static ReadCommand create(String ksName, ByteBuffer key, String cfName, long timestamp, IDiskAtomFilter filter)
{
    if (filter instanceof SliceQueryFilter)
        return new SliceFromReadCommand(ksName, key, cfName, timestamp, (SliceQueryFilter)filter);
    else
        return new SliceByNamesReadCommand(ksName, key, cfName, timestamp, (NamesQueryFilter)filter);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
public ReadCommand deserialize(DataInput in, int version) throws IOException
{
    boolean isDigest = in.readBoolean();
    String keyspaceName = in.readUTF();
    ByteBuffer key = ByteBufferUtil.readWithShortLength(in);
    String cfName = in.readUTF();
    long timestamp = in.readLong();
    CFMetaData metadata = Schema.instance.getCFMetaData(keyspaceName, cfName);
    SliceQueryFilter filter = metadata.comparator.sliceQueryFilterSerializer().deserialize(in, version);
    return new SliceFromReadCommand(keyspaceName, key, cfName, timestamp, filter).setIsDigestQuery(isDigest);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
public static List<Row> getRangeSlice(ColumnFamilyStore cfs, ByteBuffer superColumn)
{
    IDiskAtomFilter filter = superColumn == null
                           ? new IdentityQueryFilter()
                           : new SliceQueryFilter(SuperColumns.startOf(superColumn), SuperColumns.endOf(superColumn), false, Integer.MAX_VALUE);

    Token min = StorageService.getPartitioner().getMinimumToken();
    return cfs.getRangeSlice(new Bounds<Token>(min, min).toRowBounds(), null, filter, 10000);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
private void findRowGetSlicesAndAssertColsFound(ColumnFamilyStore cfs, SliceQueryFilter filter, String rowKey,
        String... colNames)
{
    List<Row> rows = cfs.getRangeSlice(new Bounds<RowPosition>(rp(rowKey), rp(rowKey)),
                                       null,
                                       filter,
                                       Integer.MAX_VALUE,
                                       System.currentTimeMillis(),
                                       false,
                                       false);
    assertSame("unexpected number of rows ", 1, rows.size());
    Row row = rows.get(0);
    Collection<Cell> cols = !filter.isReversed() ? row.cf.getSortedColumns() : row.cf.getReverseSortedColumns();
    // printRow(cfs, new String(row.key.key.array()), cols);
    String[] returnedColsNames = Iterables.toArray(Iterables.transform(cols, new Function<Cell, String>()
    {
        public String apply(Cell arg0)
        {
            return Util.string(arg0.name().toByteBuffer());
        }
    }), String.class);

    assertTrue(
            "Columns did not match. Expected: " + Arrays.toString(colNames) + " but got:"
                    + Arrays.toString(returnedColsNames), Arrays.equals(colNames, returnedColsNames));
    int i = 0;
    for (Cell col : cols)
    {
        assertEquals(colNames[i++], Util.string(col.name().toByteBuffer()));
    }
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
protected List<Row> queryNextPage(int pageSize, ConsistencyLevel consistencyLevel, boolean localQuery)
throws RequestValidationException, RequestExecutionException
{
    SliceQueryFilter filter = command.filter.withUpdatedCount(pageSize);
    if (lastReturned != null)
        filter = filter.withUpdatedStart(lastReturned, cfm.comparator);

    ReadCommand pageCmd = command.withUpdatedFilter(filter);
    return localQuery
         ? Collections.singletonList(pageCmd.getRow(Keyspace.open(command.ksName)))
         : StorageProxy.read(Collections.singletonList(pageCmd), consistencyLevel);
}
 

import org.apache.cassandra.db.filter.SliceQueryFilter; //导入依赖的package包/类
public void serialize(ReadCommand rm, DataOutput out, int version) throws IOException
{
    SliceFromReadCommand realRM = (SliceFromReadCommand)rm;
    out.writeBoolean(realRM.isDigestQuery());
    out.writeUTF(realRM.ksName);
    ByteBufferUtil.writeWithShortLength(realRM.key, out);
    out.writeUTF(realRM.cfName);
    out.writeLong(realRM.timestamp);
    SliceQueryFilter.serializer.serialize(realRM.filter, out, version);
}