import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
/**
 * Post processes in the coordinator node the results of a distributed
 * search. Gets the k globally best results from all the k best node-local
 * results.
 *
 * @param partitions
 *            the node results iterator
 * @param command
 *            the read command
 * @return the k globally best results
 */
public PartitionIterator postProcess(PartitionIterator partitions, ReadCommand command) {

	// Search search = search(command);
	//
	// // Skip if search does not require full scan
	// if (search.requiresFullScan()) {
	//
	// List<Pair<DecoratedKey, SimpleRowIterator>> collectedRows =
	// collect(partitions);
	//
	// Query query = search.query(schema);
	// Sort sort = sort(search);
	// int limit = command.limits().count();
	//
	// // Skip if search is not top-k TODO: Skip if only one partitioner
	// // range is involved
	// if (search.isTopK()) {
	// return process(query, sort, limit, collectedRows);
	// }
	// }
	return partitions;
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
/**
 * Returns the Lucene {@link Query} represented by the specified
 * {@link Search} and key filter.
 *
 * @param expression
 *            the expression
 * @param command
 *            the command
 * @return a Lucene {@link Query}
 */
private Query query(String expression, ReadCommand command) {
	try {
		QueryParser queryParser = new QueryParser("query", this.indexOptions.search.defaultAnalyzer);
		queryParser.setDateResolution(Resolution.SECOND);
		Query searchQuery = queryParser.parse(expression);
		
		return searchQuery;
	} catch (ParseException e) {
		// TODO Auto-generated catch block
		throw new FhirIndexException(e);
	}

	// TODO: mejorar las busquedas por tipo
	// Optional<Query> maybeKeyRangeQuery = query(command);
	// if (maybeKeyRangeQuery.isPresent()) {
	// BooleanQuery.Builder builder = new BooleanQuery.Builder();
	// builder.add(maybeKeyRangeQuery.get(), FILTER);
	// builder.add(searchQuery, MUST);
	// return builder.build();
	// } else {
	// }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
private void makeRequests(ReadCommand readCommand, Iterable<InetAddress> endpoints)
{
    MessageOut<ReadCommand> message = null;
    boolean hasLocalEndpoint = false;

    for (InetAddress endpoint : endpoints)
    {
        if (isLocalRequest(endpoint))
        {
            hasLocalEndpoint = true;
            continue;
        }

        logger.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
        if (message == null)
            message = readCommand.createMessage();
        MessagingService.instance().sendRR(message, endpoint, handler);
    }

    // We delay the local (potentially blocking) read till the end to avoid stalling remote requests.
    if (hasLocalEndpoint)
    {
        logger.trace("reading {} locally", readCommand.isDigestQuery() ? "digest" : "data");
        StageManager.getStage(Stage.READ).maybeExecuteImmediately(new LocalReadRunnable(command, handler));
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
public void maybeTryAdditionalReplicas()
{
    // no latency information, or we're overloaded
    if (cfs.sampleLatencyNanos > TimeUnit.MILLISECONDS.toNanos(command.getTimeout()))
        return;

    if (!handler.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS))
    {
        // Could be waiting on the data, or on enough digests.
        ReadCommand retryCommand = command;
        if (resolver.getData() != null)
            retryCommand = command.copy().setIsDigestQuery(true);

        InetAddress extraReplica = Iterables.getLast(targetReplicas);
        logger.trace("speculating read retry on {}", extraReplica);
        MessagingService.instance().sendRR(retryCommand.createMessage(), extraReplica, handler);
        speculated = true;

        cfs.metric.speculativeRetries.inc();
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
protected void makeDigestRequests(Iterable<InetAddress> endpoints)
{
    ReadCommand digestCommand = command.copy();
    digestCommand.setDigestQuery(true);
    MessageOut<?> message = digestCommand.createMessage();
    for (InetAddress endpoint : endpoints)
    {
        if (isLocalRequest(endpoint))
        {
            logger.trace("reading digest locally");
            StageManager.getStage(Stage.READ).execute(new LocalReadRunnable(digestCommand, handler));
        }
        else
        {
            logger.trace("reading digest from {}", endpoint);
            MessagingService.instance().sendRR(message, endpoint, handler);
        }
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
public void maybeTryAdditionalReplicas()
{
    // no latency information, or we're overloaded
    if (cfs.sampleLatencyNanos > TimeUnit.MILLISECONDS.toNanos(command.getTimeout()))
        return;

    if (!handler.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS))
    {
        // Could be waiting on the data, or on enough digests.
        ReadCommand retryCommand = command;
        if (resolver.getData() != null)
        {
            retryCommand = command.copy();
            retryCommand.setDigestQuery(true);
        }

        InetAddress extraReplica = Iterables.getLast(targetReplicas);
        logger.trace("speculating read retry on {}", extraReplica);
        MessagingService.instance().sendRR(retryCommand.createMessage(), extraReplica, handler);
        speculated = true;

        cfs.metric.speculativeRetries.inc();
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
private void retryDummyRead(String ks, String cf) throws PermanentBackendException {

        final long limit = System.currentTimeMillis() + (60L * 1000L);

        while (System.currentTimeMillis() < limit) {
            try {
                SortedSet<ByteBuffer> ss = new TreeSet<ByteBuffer>();
                ss.add(ByteBufferUtil.zeroByteBuffer(1));
                NamesQueryFilter nqf = new NamesQueryFilter(ss);
                SliceByNamesReadCommand cmd = new SliceByNamesReadCommand(ks, ByteBufferUtil.zeroByteBuffer(1), cf, 1L, nqf);
                StorageProxy.read(ImmutableList.<ReadCommand> of(cmd), ConsistencyLevel.QUORUM);
                log.info("Read on CF {} in KS {} succeeded", cf, ks);
                return;
            } catch (Throwable t) {
                log.warn("Failed to read CF {} in KS {} following creation", cf, ks, t);
            }

            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                throw new PermanentBackendException(e);
            }
        }

        throw new PermanentBackendException("Timed out while attempting to read CF " + cf + " in KS " + ks + " following creation");
    }
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
AbstractReadExecutor(Keyspace keyspace, ReadCommand command, ConsistencyLevel consistencyLevel, List<InetAddress> targetReplicas)
{
    this.command = command;
    this.targetReplicas = targetReplicas;
    this.handler = new ReadCallback(new DigestResolver(keyspace, command, consistencyLevel, targetReplicas.size()), consistencyLevel, command, targetReplicas);
    this.traceState = Tracing.instance.get();

    // Set the digest version (if we request some digests). This is the smallest version amongst all our target replicas since new nodes
    // knows how to produce older digest but the reverse is not true.
    // TODO: we need this when talking with pre-3.0 nodes. So if we preserve the digest format moving forward, we can get rid of this once
    // we stop being compatible with pre-3.0 nodes.
    int digestVersion = MessagingService.current_version;
    for (InetAddress replica : targetReplicas)
        digestVersion = Math.min(digestVersion, MessagingService.instance().getVersion(replica));
    command.setDigestVersion(digestVersion);
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
private void makeRequests(ReadCommand readCommand, Iterable<InetAddress> endpoints)
{
    boolean hasLocalEndpoint = false;

    for (InetAddress endpoint : endpoints)
    {
        if (StorageProxy.canDoLocalRequest(endpoint))
        {
            hasLocalEndpoint = true;
            continue;
        }

        if (traceState != null)
            traceState.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
        logger.trace("reading {} from {}", readCommand.isDigestQuery() ? "digest" : "data", endpoint);
        MessageOut<ReadCommand> message = readCommand.createMessage(MessagingService.instance().getVersion(endpoint));
        MessagingService.instance().sendRRWithFailure(message, endpoint, handler);
    }

    // We delay the local (potentially blocking) read till the end to avoid stalling remote requests.
    if (hasLocalEndpoint)
    {
        logger.trace("reading {} locally", readCommand.isDigestQuery() ? "digest" : "data");
        StageManager.getStage(Stage.READ).maybeExecuteImmediately(new LocalReadRunnable(command, handler));
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
public void maybeTryAdditionalReplicas()
{
    // no latency information, or we're overloaded
    if (cfs.sampleLatencyNanos > TimeUnit.MILLISECONDS.toNanos(command.getTimeout()))
        return;

    if (!handler.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS))
    {
        // Could be waiting on the data, or on enough digests.
        ReadCommand retryCommand = command;
        if (handler.resolver.isDataPresent())
            retryCommand = command.copy().setIsDigestQuery(true);

        InetAddress extraReplica = Iterables.getLast(targetReplicas);
        if (traceState != null)
            traceState.trace("speculating read retry on {}", extraReplica);
        logger.trace("speculating read retry on {}", extraReplica);
        int version = MessagingService.instance().getVersion(extraReplica);
        MessagingService.instance().sendRRWithFailure(retryCommand.createMessage(version), extraReplica, handler);
        speculated = true;

        cfs.metric.speculativeRetries.inc();
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
private List<ReadCommand> getSliceTravelCommands(QueryOptions options, long now, int step) throws RequestValidationException
{
	List<ReadCommand> commands = new ArrayList<>();

	Collection<ByteBuffer> keys = getKeys(options, step);
    //if (keys.isEmpty()) // in case of IN () for (the last column of) the partition key.
    
	IDiskAtomFilter filter = makeFilter(options, step);
    if (filter == null)
        return null;

    if (keys.isEmpty()){
    	//if we do not have keys yet, just create a Read Command with empty key.
    	ByteBuffer empty = ByteBuffer.wrap("EMPTY".getBytes());
    	//logger.info("@daidong debug: create read command with empty key ");
    	commands.add(ReadCommand.create(keyspace(), ByteBufferUtil.clone(empty), columnFamily(), now, filter.cloneShallow()));
    } else {
    	for (ByteBuffer key : keys)
    	{
    		QueryProcessor.validateKey(key);
    		//logger.info("@daidong debug: create read command with keys " + key.toString());
    		commands.add(ReadCommand.create(keyspace(), ByteBufferUtil.clone(key), columnFamily(), now, filter.cloneShallow()));
    	}
    }
    return commands;
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的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.ReadCommand; //导入依赖的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.ReadCommand; //导入依赖的package包/类
protected void runMayThrow() throws IOException
{
    try
    {
        resolver.resolve();
    }
    catch (DigestMismatchException e)
    {
        if (logger.isDebugEnabled())
            logger.debug("Digest mismatch:", e);

        ReadCommand readCommand = (ReadCommand) command;
        final RowRepairResolver repairResolver = new RowRepairResolver(readCommand.table, readCommand.key);
        IAsyncCallback repairHandler = new AsyncRepairCallback(repairResolver, endpoints.size());

        for (InetAddress endpoint : endpoints)
            MessagingService.instance().sendRR(readCommand, endpoint, repairHandler);
    }
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
/**
 * This method is invoked when a CQL query is executed.
 */
public Searcher searcherFor(ReadCommand command) {
	logger.trace("Getting searcher for {}", command);
	try {
		return service.searcher(command);
	} catch (Exception e) {
		logger.error("Error while searching", e);
		throw new InvalidRequestException(e.getMessage());
	}
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
/**
 * Gets the optional {@link SearchCacheEntry} associated to the specified search and {@link ReadCommand}.
 *
 * @param search the search
 * @param command the read command
 * @return the cache entry, maybe empty
 */
public Optional<SearchCacheEntry> get(String search, ReadCommand command) {
    if (command instanceof PartitionRangeReadCommand) {
        PartitionRangeReadCommand rangeCommand = (PartitionRangeReadCommand) command;
        return cache.asMap().values().stream().filter(e -> e.isValid(comparator, search, rangeCommand)).findAny();
    }
    return Optional.empty();
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
SearchCacheUpdater(SearchCache cache, String search, UUID id, ReadCommand command, Query query) {
    this.cache = cache;
    this.search = search;
    this.id = id;
    this.command = command instanceof PartitionRangeReadCommand ? (PartitionRangeReadCommand) command : null;
    this.query = query;
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
private void retryDummyRead(String ks, String cf) throws PermanentBackendException {

        final long limit = System.currentTimeMillis() + (60L * 1000L);

        while (System.currentTimeMillis() < limit) {
            try {
                SortedSet<CellName> names = new TreeSet<>(new Comparator<CellName>() {
                    // This is a singleton set.  We need to define a comparator because SimpleDenseCellName is not
                    // comparable, but it doesn't have to be a useful comparator
                    @Override
                    public int compare(CellName o1, CellName o2)
                    {
                        return 0;
                    }
                });
                names.add(CellNames.simpleDense(ByteBufferUtil.zeroByteBuffer(1)));
                NamesQueryFilter nqf = new NamesQueryFilter(names);
                SliceByNamesReadCommand cmd = new SliceByNamesReadCommand(ks, ByteBufferUtil.zeroByteBuffer(1), cf, 1L, nqf);
                StorageProxy.read(ImmutableList.<ReadCommand> of(cmd), ConsistencyLevel.QUORUM);
                log.info("Read on CF {} in KS {} succeeded", cf, ks);
                return;
            } catch (Throwable t) {
                log.warn("Failed to read CF {} in KS {} following creation", cf, ks, t);
            }

            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                throw new PermanentBackendException(e);
            }
        }

        throw new PermanentBackendException("Timed out while attempting to read CF " + cf + " in KS " + ks + " following creation");
    }
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
AbstractReadExecutor(ReadCommand command, ConsistencyLevel consistencyLevel, List<InetAddress> targetReplicas)
{
    this.command = command;
    this.targetReplicas = targetReplicas;
    resolver = new RowDigestResolver(command.ksName, command.key);
    handler = new ReadCallback<>(resolver, consistencyLevel, command, targetReplicas);
}
 

import org.apache.cassandra.db.ReadCommand; //导入依赖的package包/类
public SpeculatingReadExecutor(ColumnFamilyStore cfs,
                               ReadCommand command,
                               ConsistencyLevel consistencyLevel,
                               List<InetAddress> targetReplicas)
{
    super(command, consistencyLevel, targetReplicas);
    this.cfs = cfs;
}