import org.apache.cassandra.db.Row; //导入依赖的package包/类
public Iterable<Row> resolve()
{
    ArrayList<RowIterator> iters = new ArrayList<RowIterator>(responses.size());
    int n = 0;
    for (MessageIn<RangeSliceReply> response : responses)
    {
        RangeSliceReply reply = response.payload;
        n = Math.max(n, reply.rows.size());
        iters.add(new RowIterator(reply.rows.iterator(), response.from));
    }
    // for each row, compute the combination of all different versions seen, and repair incomplete versions
    // TODO do we need to call close?
    CloseableIterator<Row> iter = MergeIterator.get(iters, pairComparator, new Reducer());

    List<Row> resolvedRows = new ArrayList<Row>(n);
    while (iter.hasNext())
        resolvedRows.add(iter.next());

    return resolvedRows;
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
protected Row getReduced()
{
    ColumnFamily resolved = versions.size() > 1
                          ? RowDataResolver.resolveSuperset(versions, timestamp)
                          : versions.get(0);
    if (versions.size() < sources.size())
    {
        // add placeholder rows for sources that didn't have any data, so maybeScheduleRepairs sees them
        for (InetAddress source : sources)
        {
            if (!versionSources.contains(source))
            {
                versions.add(null);
                versionSources.add(source);
            }
        }
    }
    // resolved can be null even if versions doesn't have all nulls because of the call to removeDeleted in resolveSuperSet
    if (resolved != null)
        repairResults.addAll(RowDataResolver.scheduleRepairs(resolved, keyspaceName, key, versions, versionSources));
    versions.clear();
    versionSources.clear();
    return new Row(key, resolved);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * Performs a search across a number of column indexes
 *
 * @param filter the column range to restrict to
 * @return found indexed rows
 */
public List<Row> search(ExtendedFilter filter)
{
    List<SecondaryIndexSearcher> indexSearchers = getIndexSearchersForQuery(filter.getClause());

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

    SecondaryIndexSearcher mostSelective = null;
    long bestEstimate = Long.MAX_VALUE;
    for (SecondaryIndexSearcher searcher : indexSearchers)
    {
        SecondaryIndex highestSelectivityIndex = searcher.highestSelectivityIndex(filter.getClause());
        long estimate = highestSelectivityIndex.estimateResultRows();
        if (estimate <= bestEstimate)
        {
            bestEstimate = estimate;
            mostSelective = searcher;
        }
    }

    return mostSelective.search(filter);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
public void reload()
{
    Row cfDefRow = SystemKeyspace.readSchemaRow(SystemKeyspace.SCHEMA_COLUMNFAMILIES_CF, ksName, cfName);

    if (cfDefRow.cf == null || !cfDefRow.cf.hasColumns())
        throw new RuntimeException(String.format("%s not found in the schema definitions keyspace.", ksName + ":" + cfName));

    try
    {
        apply(fromSchema(cfDefRow));
    }
    catch (ConfigurationException e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
@Override
protected SecondaryIndexSearcher createSecondaryIndexSearcher(Set<ByteBuffer> columns)
{
    return new SecondaryIndexSearcher(baseCfs.indexManager, columns)
    {
        
        @Override
        public List<Row> search(ExtendedFilter filter)
        {
            return Arrays.asList(new Row(LAST_INDEXED_KEY, LAST_INDEXED_ROW));
        }

        @Override
        public void validate(IndexExpression indexExpression) throws InvalidRequestException
        {
            if (indexExpression.value.equals(ByteBufferUtil.bytes("invalid")))
                throw new InvalidRequestException("Invalid search!");
        }
        
    };
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
public static ColumnFamily remoteStorageQuery(String target, String dataTag){
	try {
		List<ReadCommand> command = new ArrayList<ReadCommand>();
		command.add(new SliceFromReadCommand(
				Metadata.MetaData_KS,
				ByteBufferUtil.bytes(target), 
				new QueryPath(Metadata.MetadataLog_CF),
				Column.decomposeName(String.valueOf(0) , "", dataTag, "value"),
				Column.decomposeName(String.valueOf(Long.MAX_VALUE) , "", dataTag, "value"),
				false,
				Integer.MAX_VALUE));

		List<Row> rows = StorageProxy.read(command, ConsistencyLevel.ANY);
		return rows.get(0).cf;
		 
	} catch (Exception e) {
		return null;
	}
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
private ColumnFamily remoteStorageQuery(String target, String dataTag){
	try {
		List<ReadCommand> command = new ArrayList<ReadCommand>();
		command.add(new SliceFromReadCommand(
				Metadata.MetaData_KS,
				ByteBufferUtil.bytes(target), 
				new QueryPath(Metadata.MetadataRegistry_CF),
				Column.decomposeName(dataTag, "admin_tag"),
				Column.decomposeName(dataTag, "admin_tag"),
				false,
				Integer.MAX_VALUE));

		List<Row> rows = StorageProxy.read(command, ConsistencyLevel.ANY);
		return rows.get(0).cf;
		 
	} catch (Exception e) {
		return null;
	}
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * Special case of resolve() so that CL.ONE reads never throw DigestMismatchException in the foreground
 */
public Row getData()
{
    for (MessageIn<ReadResponse> message : replies)
    {
        ReadResponse result = message.payload;
        if (!result.isDigestQuery())
            return result.row();
    }
    return null;
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
public Row resolve() throws DigestMismatchException
{
    if (logger.isDebugEnabled())
        logger.debug("resolving {} responses", replies.size());

    long start = System.nanoTime();

    // validate digests against each other; throw immediately on mismatch.
    // also extract the data reply, if any.
    ColumnFamily data = null;
    ByteBuffer digest = null;

    for (MessageIn<ReadResponse> message : replies)
    {
        ReadResponse response = message.payload;

        ByteBuffer newDigest;
        if (response.isDigestQuery())
        {
            newDigest = response.digest();
        }
        else
        {
            // note that this allows for multiple data replies, post-CASSANDRA-5932
            data = response.row().cf;
            newDigest = ColumnFamily.digest(data);
        }

        if (digest == null)
            digest = newDigest;
        else if (!digest.equals(newDigest))
            throw new DigestMismatchException(key, digest, newDigest);
    }

    if (logger.isDebugEnabled())
        logger.debug("resolve: {} ms.", TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
    return new Row(key, data);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
@Override
public List<Row> search(ExtendedFilter filter)
{
    assert filter.getClause() != null && !filter.getClause().isEmpty();
    final IndexExpression primary = highestSelectivityPredicate(filter.getClause());
    final CompositesIndex index = (CompositesIndex)indexManager.getIndexForColumn(primary.column);
    // TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try* to delete stale entries, without blocking if there's no room
    // as it stands, we open a writeOp and keep it open for the duration to ensure that should this CF get flushed to make room we don't block the reclamation of any room being made
    try (OpOrder.Group writeOp = baseCfs.keyspace.writeOrder.start(); OpOrder.Group baseOp = baseCfs.readOrdering.start(); OpOrder.Group indexOp = index.getIndexCfs().readOrdering.start())
    {
        return baseCfs.filter(getIndexedIterator(writeOp, filter, primary, index), filter);
    }
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
private ResultMessage.Rows execute(Pageable command, QueryOptions options, int limit, long now) throws RequestValidationException, RequestExecutionException
{
    List<Row> trows;
    if (command == null)
        trows = Collections.<Row>emptyList();
    else
    	trows = StorageProxy.travel(((TravelCommand) command), options.getConsistency());        	
    return processResults(trows, options, limit, now);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
public ResultMessage.Rows processResults(List<Row> rows, QueryOptions options, int limit, long now) throws RequestValidationException
{
    // Even for count, we need to process the result as it'll group some column together in sparse column families
    ResultSet rset = process(rows, options, limit, now);
    ResultMessage.Rows rtn = new ResultMessage.Rows(rset);
    //logger.info("@daidong debug: get results: " + rtn.toString());
    return rtn;
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
private ResultSet process(List<Row> rows, QueryOptions options, int limit, long now) throws InvalidRequestException
{
    Selection.ResultSetBuilder result = selection.resultSetBuilder(now);
    for (org.apache.cassandra.db.Row row : rows)
    {
        // Not columns match the query, skip
        if (row.cf == null)
            continue;

        processColumnFamily(row.key.getKey(), row.cf, options, now, result);
    }
    ResultSet cqlRows = result.build();
    
    return cqlRows;
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * Create CFMetaData from thrift {@link CqlRow} that contains columns from schema_columnfamilies.
 *
 * @param columnsRes CqlRow containing columns from schema_columnfamilies.
 * @return CFMetaData derived from CqlRow
 */
public static CFMetaData fromThriftCqlRow(CqlRow cf, CqlResult columnsRes)
{
    UntypedResultSet.Row cfRow = new UntypedResultSet.Row(convertThriftCqlRow(cf));

    List<Map<String, ByteBuffer>> cols = new ArrayList<>(columnsRes.rows.size());
    for (CqlRow row : columnsRes.rows)
        cols.add(convertThriftCqlRow(row));
    UntypedResultSet colsRow = UntypedResultSet.create(cols);

    return fromSchemaNoTriggers(cfRow, colsRow);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * Deserialize CF metadata from low-level representation
 *
 * @return Thrift-based metadata deserialized from schema
 */
public static CFMetaData fromSchema(UntypedResultSet.Row result)
{
    String ksName = result.getString("keyspace_name");
    String cfName = result.getString("columnfamily_name");

    Row serializedColumns = SystemKeyspace.readSchemaRow(SystemKeyspace.SCHEMA_COLUMNS_CF, ksName, cfName);
    CFMetaData cfm = fromSchemaNoTriggers(result, ColumnDefinition.resultify(serializedColumns));

    Row serializedTriggers = SystemKeyspace.readSchemaRow(SystemKeyspace.SCHEMA_TRIGGERS_CF, ksName, cfName);
    addTriggerDefinitionsFromSchema(cfm, serializedTriggers);

    return cfm;
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
public static List<Row> executeLocally(RangeSliceCommand command) throws ExecutionException, InterruptedException
{
    ColumnFamilyStore cfs = Table.open(command.keyspace).getColumnFamilyStore(command.column_family);
    if (cfs.indexManager.hasIndexFor(command.row_filter))
        return cfs.search(command.row_filter, command.range, command.maxResults, command.predicate, command.countCQL3Rows);
    else
        return cfs.getRangeSlice(command.super_column, command.range, command.maxResults, command.predicate, command.row_filter, command.countCQL3Rows, command.isPaging);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * Special case of resolve() so that CL.ONE reads never throw DigestMismatchException in the foreground
 */
public Row getData()
{
    for (MessageIn<ReadResponse> message : replies)
    {
        ReadResponse result = message.payload;
        if (!result.isDigestQuery())
            return result.row();
    }

    throw new AssertionError("getData should not be invoked when no data is present");
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
@Override
public List<Row> search(ExtendedFilter filter)
{
    assert filter.getClause() != null && !filter.getClause().isEmpty();
    final IndexExpression primary = highestSelectivityPredicate(filter.getClause(), true);
    final CompositesIndex index = (CompositesIndex)indexManager.getIndexForColumn(primary.column);
    // TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try* to delete stale entries, without blocking if there's no room
    // as it stands, we open a writeOp and keep it open for the duration to ensure that should this CF get flushed to make room we don't block the reclamation of any room being made
    try (OpOrder.Group writeOp = baseCfs.keyspace.writeOrder.start(); OpOrder.Group baseOp = baseCfs.readOrdering.start(); OpOrder.Group indexOp = index.getIndexCfs().readOrdering.start())
    {
        return baseCfs.filter(getIndexedIterator(writeOp, filter, primary, index), filter);
    }
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * Performs a search across a number of column indexes
 *
 * @param filter the column range to restrict to
 * @return found indexed rows
 */
public List<Row> search(ExtendedFilter filter)
{
    SecondaryIndexSearcher mostSelective = getHighestSelectivityIndexSearcher(filter.getClause());
    if (mostSelective == null)
        return Collections.emptyList();
    else
        return mostSelective.search(filter);
}
 

import org.apache.cassandra.db.Row; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public int compare(Row row1, Row row2) {
    Float score1 = rowService.score(row1);
    Float score2 = rowService.score(row2);
    return score2.compareTo(score1);
}