Java 类org.apache.cassandra.thrift.IndexClause 实例源码

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;
}

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;
}

public IndexScanCommand(String keyspace, String column_family, IndexClause index_clause, SlicePredicate predicate, AbstractBounds<RowPosition> range)
{

    this.keyspace = keyspace;
    this.column_family = column_family;
    this.index_clause = index_clause;
    this.predicate = predicate;
    this.range = range;
}

public IndexScanCommand(String keyspace, String column_family, IndexClause index_clause, SlicePredicate predicate, AbstractBounds<RowPosition> range)
{

    this.keyspace = keyspace;
    this.column_family = column_family;
    this.index_clause = index_clause;
    this.predicate = predicate;
    this.range = range;
}

public IndexScanCommand(String keyspace, String column_family, IndexClause index_clause, SlicePredicate predicate, AbstractBounds<RowPosition> range)
{

    this.keyspace = keyspace;
    this.column_family = column_family;
    this.index_clause = index_clause;
    this.predicate = predicate;
    this.range = range;
}

private NamesQueryFilter getExtraFilter(IndexClause clause)
{
    SortedSet<ByteBuffer> columns = new TreeSet<ByteBuffer>(getComparator());
    for (IndexExpression expr : clause.expressions)
    {
        columns.add(expr.column_name);
    }
    return new NamesQueryFilter(columns);
}

public IndexScanCommand(String keyspace, String column_family, IndexClause index_clause, SlicePredicate predicate, AbstractBounds range)
{

    this.keyspace = keyspace;
    this.column_family = column_family;
    this.index_clause = index_clause;
    this.predicate = predicate;
    this.range = range;
}

public IndexScanCommand deserialize(DataInput in) throws IOException
{
    String keyspace = in.readUTF();
    String columnFamily = in.readUTF();

    TDeserializer dser = new TDeserializer(new TBinaryProtocol.Factory());
    IndexClause indexClause = new IndexClause();
    FBUtilities.deserialize(dser, indexClause, in);
    SlicePredicate predicate = new SlicePredicate();
    FBUtilities.deserialize(dser, predicate, in);
    AbstractBounds range = AbstractBounds.serializer().deserialize(in);

    return new IndexScanCommand(keyspace, columnFamily, indexClause, predicate, range);
}

public static List<Row> scan(final String keyspace, String column_family, IndexClause index_clause, SlicePredicate column_predicate, ConsistencyLevel consistency_level)
throws IOException, TimeoutException, UnavailableException
{
    IPartitioner p = StorageService.getPartitioner();

    Token leftToken = index_clause.start_key == null ? p.getMinimumToken() : p.getToken(index_clause.start_key);
    List<AbstractBounds> ranges = getRestrictedRanges(new Bounds(leftToken, p.getMinimumToken()));
    logger.debug("scan ranges are " + StringUtils.join(ranges, ","));

    // now scan until we have enough results
    List<Row> rows = new ArrayList<Row>(index_clause.count);
    for (AbstractBounds range : ranges)
    {
        List<InetAddress> liveEndpoints = StorageService.instance.getLiveNaturalEndpoints(keyspace, range.right);
        DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), liveEndpoints);

        // collect replies and resolve according to consistency level
        RangeSliceResponseResolver resolver = new RangeSliceResponseResolver(keyspace, liveEndpoints);
        IReadCommand iCommand = new IReadCommand()
        {
            public String getKeyspace()
            {
                return keyspace;
            }
        };
        ReadCallback<Iterable<Row>> handler = getReadCallback(resolver, iCommand, consistency_level, liveEndpoints);
        handler.assureSufficientLiveNodes();

        IndexScanCommand command = new IndexScanCommand(keyspace, column_family, index_clause, column_predicate, range);
        MessageProducer producer = new CachingMessageProducer(command);
        for (InetAddress endpoint : liveEndpoints)
        {
            MessagingService.instance().sendRR(producer, endpoint, handler);
            if (logger.isDebugEnabled())
                logger.debug("reading " + command + " from " + endpoint);
        }

        try
        {
            for (Row row : handler.get())
            {
                rows.add(row);
                logger.debug("read {}", row);
            }
        }
        catch (TimeoutException ex)
        {
            if (logger.isDebugEnabled())
                logger.debug("Index scan timeout: {}", ex.toString());
            throw ex;
        }
        catch (DigestMismatchException e)
        {
            throw new RuntimeException(e);
        }
        if (rows.size() >= index_clause.count)
            return rows.subList(0, index_clause.count);
    }

    return rows;
}