import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
@Test
public void testSatisfiedByWithMultipleTerms()
{
    final ByteBuffer comment = UTF8Type.instance.decompose("comment");
    final ColumnFamilyStore store = Keyspace.open("sasecondaryindex").getColumnFamilyStore("saindexed1");
    final IPartitioner<?> partitioner = StorageService.getPartitioner();

    ColumnFamily cf = ArrayBackedSortedColumns.factory.create(store.metadata);
    cf.addColumn(new Column(comment, UTF8Type.instance.decompose("software engineer is working on a project"), System.currentTimeMillis()));

    Operation.Builder builder = new Operation.Builder(OperationType.AND, UTF8Type.instance, controller,
                                        new IndexExpression(comment, IndexOperator.EQ, UTF8Type.instance.decompose("eng is a work")));
    Operation op = builder.complete();

    Assert.assertTrue(op.satisfiedBy(new Row(partitioner.decorateKey(UTF8Type.instance.decompose("key1")), cf), null, false));

    builder = new Operation.Builder(OperationType.AND, UTF8Type.instance, controller,
                                        new IndexExpression(comment, IndexOperator.EQ, UTF8Type.instance.decompose("soft works fine")));
    op = builder.complete();

    Assert.assertTrue(op.satisfiedBy(new Row(partitioner.decorateKey(UTF8Type.instance.decompose("key1")), cf), null, false));
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
private RangeSliceCommand getRangeCommand(List<ByteBuffer> variables) throws RequestValidationException
{
    IDiskAtomFilter filter = makeFilter(variables);
    if (filter == null)
        return null;

    List<IndexExpression> expressions = getIndexExpressions(variables);
    // The LIMIT provided by the user is the number of CQL row he wants returned.
    // We want to have getRangeSlice to count the number of columns, not the number of keys.
    AbstractBounds<RowPosition> keyBounds = getKeyBounds(variables);
    return keyBounds == null
         ? null
         : new RangeSliceCommand(keyspace(),
                                 columnFamily(),
                                 null,
                                 filter,
                                 keyBounds,
                                 expressions,
                                 getLimit(),
                                 true,
                                 false);
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
/**
 * @return true if the indexes can handle the clause.
 */
public boolean hasIndexFor(List<IndexExpression> clause)
{
    if (clause == null || clause.isEmpty())
        return false;

    // It doesn't seem a clause can have multiple searchers, but since
    // getIndexSearchersForQuery returns a list ...
    List<SecondaryIndexSearcher> searchers = getIndexSearchersForQuery(clause);
    if (searchers.isEmpty())
        return false;

    for (SecondaryIndexSearcher searcher : searchers)
        if (!searcher.isIndexing(clause))
            return false;
    return true;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
private boolean needsExtraQuery(ColumnFamily data)
{
    if (!(originalFilter instanceof SliceQueryFilter))
        return false;

    SliceQueryFilter filter = (SliceQueryFilter)originalFilter;
    // Check if we've fetch the whole row
    if (filter.slices.length == 1
     && filter.start().equals(ByteBufferUtil.EMPTY_BYTE_BUFFER)
     && filter.finish().equals(ByteBufferUtil.EMPTY_BYTE_BUFFER)
     && filter.count == Integer.MAX_VALUE)
        return false;

    for (IndexExpression expr : clause)
    {
        if (data.getColumn(expr.column_name) == null)
        {
            logger.debug("adding extraFilter to cover additional expressions");
            return true;
        }
    }
    return false;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public boolean isSatisfiedBy(ColumnFamily data, ColumnNameBuilder builder)
{
    // We enforces even the primary clause because reads are not synchronized with writes and it is thus possible to have a race
    // where the index returned a row which doesn't have the primary column when we actually read it
    for (IndexExpression expression : clause)
    {
        // check column data vs expression
        ByteBuffer colName = builder == null ? expression.column_name : builder.copy().add(expression.column_name).build();
        IColumn column = data.getColumn(colName);
        if (column == null)
            return false;
        int v = data.metadata().getValueValidator(expression.column_name).compare(column.value(), expression.value);
        if (!satisfies(v, expression.op))
            return false;
    }
    return true;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
private IndexExpression highestSelectivityPredicate(IndexClause clause)
{
    IndexExpression best = null;
    int bestMeanCount = Integer.MAX_VALUE;
    for (IndexExpression expression : clause.expressions)
    {
        ColumnFamilyStore cfs = getIndexedColumnFamilyStore(expression.column_name);
        if (cfs == null || !expression.op.equals(IndexOperator.EQ))
            continue;
        int columns = cfs.getMeanColumns();
        if (columns < bestMeanCount)
        {
            best = expression;
            bestMeanCount = columns;
        }
    }
    return best;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
private static boolean satisfies(ColumnFamily data, IndexClause clause, IndexExpression first)
{
    // We enforces even the primary clause because reads are not synchronized with writes and it is thus possible to have a race
    // where the index returned a row which doesn't have the primarycolumn when we actually read it
    for (IndexExpression expression : clause.expressions)
    {
        // check column data vs expression
        IColumn column = data.getColumn(expression.column_name);
        if (column == null)
            return false;
        int v = data.metadata().getValueValidator(expression.column_name).compare(column.value(), expression.value);
        if (!satisfies(v, expression.op))
            return false;
    }
    return true;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public boolean isIndexing(List<IndexExpression> clause)
{
    // this is a bit weak, currently just checks for the success of one column, not all
    // however, parent SIS.isIndexing only cares if one predicate is covered ... grrrr!!
    for (IndexExpression expression : clause)
    {
        if (indexedColumns.containsKey(expression.column_name))
            return true;
    }
    return false;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public Builder(OperationType operation, AbstractType<?> comparator, QueryController controller, IndexExpression... columns)
{
    this.op = operation;
    this.comparator = comparator;
    this.controller = controller;
    this.expressions = new ArrayList<>();
    Collections.addAll(expressions, columns);
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
private List<IndexExpression> getIndexExpressions(List<ByteBuffer> variables) throws InvalidRequestException
{
    if (metadataRestrictions.isEmpty())
        return Collections.<IndexExpression>emptyList();

    List<IndexExpression> expressions = new ArrayList<IndexExpression>();
    for (Map.Entry<CFDefinition.Name, Restriction> entry : metadataRestrictions.entrySet())
    {
        CFDefinition.Name name = entry.getKey();
        Restriction restriction = entry.getValue();
        if (restriction.isEquality())
        {
            assert restriction.eqValues.size() == 1; // IN is not supported for indexed columns.
            ByteBuffer value = restriction.eqValues.get(0).bindAndGet(variables);
            if (value == null)
                throw new InvalidRequestException(String.format("Unsupported null value for indexed column %s", name));
            if (value.remaining() > 0xFFFF)
                throw new InvalidRequestException("Index expression values may not be larger than 64K");
            expressions.add(new IndexExpression(name.name.key, IndexOperator.EQ, value));
        }
        else
        {
            for (Bound b : Bound.values())
            {
                if (restriction.bound(b) != null)
                {
                    ByteBuffer value = restriction.bound(b).bindAndGet(variables);
                    if (value == null)
                        throw new InvalidRequestException(String.format("Unsupported null value for indexed column %s", name));
                    if (value.remaining() > 0xFFFF)
                        throw new InvalidRequestException("Index expression values may not be larger than 64K");
                    expressions.add(new IndexExpression(name.name.key, restriction.getIndexOperator(b), value));
                }
            }
        }
    }
    return expressions;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public RangeSliceCommand(String keyspace, String column_family, ByteBuffer super_column, IDiskAtomFilter predicate, AbstractBounds<RowPosition> range, List<IndexExpression> row_filter, int maxResults, boolean countCQL3Rows, boolean isPaging)
{
    this.keyspace = keyspace;
    this.column_family = column_family;
    this.super_column = super_column;
    this.predicate = predicate;
    this.range = range;
    this.row_filter = row_filter;
    this.maxResults = maxResults;
    this.countCQL3Rows = countCQL3Rows;
    this.isPaging = isPaging;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public String expressionString(IndexExpression expr)
{
    return String.format("'%s.%s %s %s'",
                         baseCfs.columnFamily,
                         getExpressionComparator().getString(expr.column_name),
                         expr.op,
                         baseCfs.metadata.getColumn_metadata().get(expr.column_name).getValidator().getString(expr.value));
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
@Override
public List<Row> search(List<IndexExpression> clause, AbstractBounds<RowPosition> range, int maxResults, IDiskAtomFilter dataFilter, boolean countCQL3Rows)
{
    assert clause != null && !clause.isEmpty();
    ExtendedFilter filter = ExtendedFilter.create(baseCfs, dataFilter, clause, maxResults, countCQL3Rows, false);
    return baseCfs.filter(getIndexedIterator(range, filter), filter);
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
/**
 * Get a list of IndexSearchers from the union of expression index types
 * @param clause the query clause
 * @return the searchers needed to query the index
 */
private List<SecondaryIndexSearcher> getIndexSearchersForQuery(List<IndexExpression> clause)
{
    Map<String, Set<ByteBuffer>> groupByIndexType = new HashMap<String, Set<ByteBuffer>>();

    //Group columns by type
    for (IndexExpression ix : clause)
    {
        SecondaryIndex index = getIndexForColumn(ix.column_name);

        if (index == null)
            continue;

        Set<ByteBuffer> columns = groupByIndexType.get(index.getClass().getCanonicalName());

        if (columns == null)
        {
            columns = new HashSet<ByteBuffer>();
            groupByIndexType.put(index.getClass().getCanonicalName(), columns);
        }

        columns.add(ix.column_name);
    }

    List<SecondaryIndexSearcher> indexSearchers = new ArrayList<SecondaryIndexSearcher>(groupByIndexType.size());

    //create searcher per type
    for (Set<ByteBuffer> column : groupByIndexType.values())
        indexSearchers.add(getIndexForColumn(column.iterator().next()).createSecondaryIndexSearcher(column));

    return indexSearchers;
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
/**
 * Performs a search across a number of column indexes
 * TODO: add support for querying across index types
 *
 * @param clause the index query clause
 * @param range the row range to restrict to
 * @param dataFilter the column range to restrict to
 * @return found indexed rows
 */
public List<Row> search(List<IndexExpression> clause, AbstractBounds<RowPosition> range, int maxResults, IDiskAtomFilter dataFilter, boolean countCQL3Rows)
{
    List<SecondaryIndexSearcher> indexSearchers = getIndexSearchersForQuery(clause);

    if (indexSearchers.isEmpty())
        return Collections.emptyList();

    //We currently don't support searching across multiple index types
    if (indexSearchers.size() > 1)
        throw new RuntimeException("Unable to search across multiple secondary index types");


    return indexSearchers.get(0).search(clause, range, maxResults, dataFilter, countCQL3Rows);
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public static ExtendedFilter create(ColumnFamilyStore cfs, IDiskAtomFilter filter, List<IndexExpression> clause, int maxResults, boolean countCQL3Rows, boolean isPaging)
{
    if (clause == null || clause.isEmpty())
    {
        return new EmptyClauseFilter(cfs, filter, maxResults, countCQL3Rows, isPaging);
    }
    else
    {
        if (isPaging)
            throw new IllegalArgumentException("Cross-row paging is not supported along with index clauses");
        return cfs.getComparator() instanceof CompositeType
             ? new FilterWithCompositeClauses(cfs, filter, clause, maxResults, countCQL3Rows)
             : new FilterWithClauses(cfs, filter, clause, maxResults, countCQL3Rows);
    }
}
 

import org.apache.cassandra.thrift.IndexExpression; //导入依赖的package包/类
public FilterWithClauses(ColumnFamilyStore cfs, IDiskAtomFilter filter, List<IndexExpression> clause, int maxResults, boolean countCQL3Rows)
{
    super(cfs, filter, maxResults, countCQL3Rows, false);
    assert clause != null;
    this.clause = clause;
    this.initialFilter = computeInitialFilter();
}