import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage processPrepared(CQLStatement statement, QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
    List<ByteBuffer> variables = options.getValues();
    // Check to see if there are any bound variables to verify
    if (!(variables.isEmpty() && (statement.getBoundTerms() == 0)))
    {
        if (variables.size() != statement.getBoundTerms())
            throw new InvalidRequestException(String.format("there were %d markers(?) in CQL but %d bound variables",
                                                            statement.getBoundTerms(),
                                                            variables.size()));

        // at this point there is a match in count between markers and variables that is non-zero

        if (logger.isTraceEnabled())
            for (int i = 0; i < variables.size(); i++)
                logger.trace("[{}] '{}'", i+1, variables.get(i));
    }

    metrics.preparedStatementsExecuted.inc();
    return processStatement(statement, queryState, options);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
@Override
public Response execute(QueryState queryState)
{
    try
    {
        IAuthenticator.SaslNegotiator negotiator = ((ServerConnection) connection).getSaslNegotiator(queryState);
        byte[] challenge = negotiator.evaluateResponse(token);
        if (negotiator.isComplete())
        {
            AuthenticatedUser user = negotiator.getAuthenticatedUser();
            queryState.getClientState().login(user);
            // authentication is complete, send a ready message to the client
            return new AuthSuccess(challenge);
        }
        else
        {
            return new AuthChallenge(challenge);
        }
    }
    catch (AuthenticationException e)
    {
        return ErrorMessage.fromException(e);
    }
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage executeWithCondition(QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
    List<ByteBuffer> keys = buildPartitionKeyNames(options);
    // We don't support IN for CAS operation so far
    if (keys.size() > 1)
        throw new InvalidRequestException("IN on the partition key is not supported with conditional updates");

    ByteBuffer key = keys.get(0);
    long now = options.getTimestamp(queryState);
    Composite prefix = createClusteringPrefix(options);

    CQL3CasRequest request = new CQL3CasRequest(cfm, key, false);
    addConditions(prefix, request, options);
    request.addRowUpdate(prefix, this, options, now);

    ColumnFamily result = StorageProxy.cas(keyspace(),
                                           columnFamily(),
                                           key,
                                           request,
                                           options.getSerialConsistency(),
                                           options.getConsistency(),
                                           queryState.getClientState());
    return new ResultMessage.Rows(buildCasResultSet(key, result, options));
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
private CQL3CasRequest makeCasRequest(QueryState queryState, QueryOptions options)
{
    List<ByteBuffer> keys = buildPartitionKeyNames(options);
    // We don't support IN for CAS operation so far
    checkFalse(keys.size() > 1,
               "IN on the partition key is not supported with conditional %s",
               type.isUpdate()? "updates" : "deletions");

    DecoratedKey key = cfm.decorateKey(keys.get(0));
    long now = options.getTimestamp(queryState);
    SortedSet<Clustering> clusterings = createClustering(options);

    checkFalse(clusterings.size() > 1,
               "IN on the clustering key columns is not supported with conditional %s",
                type.isUpdate()? "updates" : "deletions");

    Clustering clustering = Iterables.getOnlyElement(clusterings);

    CQL3CasRequest request = new CQL3CasRequest(cfm, key, false, conditionColumns(), updatesRegularRows(), updatesStaticRow());

    addConditions(clustering, request, options);
    request.addRowUpdate(clustering, this, options, now);

    return request;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
private ResultMessage.Rows execute(Pageable command, QueryOptions options, int limit, long now, QueryState state) throws RequestValidationException, RequestExecutionException
{
    List<Row> rows;
    if (command == null)
    {
        rows = Collections.<Row>emptyList();
    }
    else
    {
        rows = command instanceof Pageable.ReadCommands
             ? StorageProxy.read(((Pageable.ReadCommands)command).commands, options.getConsistency(), state.getClientState())
             : StorageProxy.getRangeSlice((RangeSliceCommand)command, options.getConsistency());
    }

    return processResults(rows, options, limit, now);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public QueryState validateNewMessage(Message.Type type, int version, int streamId)
{
    switch (state)
    {
        case UNINITIALIZED:
            if (type != Message.Type.STARTUP && type != Message.Type.OPTIONS)
                throw new ProtocolException(String.format("Unexpected message %s, expecting STARTUP or OPTIONS", type));
            break;
        case AUTHENTICATION:
            // Support both SASL auth from protocol v2 and the older style Credentials auth from v1
            if (type != Message.Type.AUTH_RESPONSE && type != Message.Type.CREDENTIALS)
                throw new ProtocolException(String.format("Unexpected message %s, expecting %s", type, version == 1 ? "CREDENTIALS" : "SASL_RESPONSE"));
            break;
        case READY:
            if (type == Message.Type.STARTUP)
                throw new ProtocolException("Unexpected message STARTUP, the connection is already initialized");
            break;
        default:
            throw new AssertionError();
    }
    return getQueryState(streamId);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public Message.Response execute(QueryState state)
{
    List<String> cqlVersions = new ArrayList<String>();
    cqlVersions.add(QueryProcessor.CQL_VERSION.toString());

    List<String> compressions = new ArrayList<String>();
    if (FrameCompressor.SnappyCompressor.instance != null)
        compressions.add("snappy");
    // LZ4 is always available since worst case scenario it default to a pure JAVA implem.
    compressions.add("lz4");

    Map<String, List<String>> supported = new HashMap<String, List<String>>();
    supported.put(StartupMessage.CQL_VERSION, cqlVersions);
    supported.put(StartupMessage.COMPRESSION, compressions);

    return new SupportedMessage(supported);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
@Override
public Response execute(QueryState queryState)
{
    try
    {
        SaslAuthenticator authenticator = ((ServerConnection) connection).getAuthenticator();
        byte[] challenge = authenticator.evaluateResponse(token == null ? new byte[0] : token);
        if (authenticator.isComplete())
        {
            AuthenticatedUser user = authenticator.getAuthenticatedUser();
            queryState.getClientState().login(user);
            // authentication is complete, send a ready message to the client
            return new AuthSuccess(challenge);
        }
        else
        {
            return new AuthChallenge(challenge);
        }
    }
    catch (AuthenticationException e)
    {
        return ErrorMessage.fromException(e);
    }
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
static RowIterator casInternal(CQL3CasRequest request, QueryState state)
{
    UUID ballot = UUIDGen.getTimeUUIDFromMicros(state.getTimestamp());

    SinglePartitionReadCommand readCommand = request.readCommand(FBUtilities.nowInSeconds());
    FilteredPartition current;
    try (ReadOrderGroup orderGroup = readCommand.startOrderGroup(); PartitionIterator iter = readCommand.executeInternal(orderGroup))
    {
        current = FilteredPartition.create(PartitionIterators.getOnlyElement(iter, readCommand));
    }

    if (!request.appliesTo(current))
        return current.rowIterator();

    PartitionUpdate updates = request.makeUpdates(current);
    updates = TriggerExecutor.instance.execute(updates);

    Commit proposal = Commit.newProposal(ballot, updates);
    proposal.makeMutation().apply();
    return null;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage executeWithCondition(QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
    CQL3CasRequest request = makeCasRequest(queryState, options);

    try (RowIterator result = StorageProxy.cas(keyspace(),
                                               columnFamily(),
                                               request.key,
                                               request,
                                               options.getSerialConsistency(),
                                               options.getConsistency(),
                                               queryState.getClientState()))
    {
        return new ResultMessage.Rows(buildCasResultSet(result, options));
    }
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage processBatch(BatchStatement statement,
                                  QueryState state,
                                  BatchQueryOptions options,
                                  Map<String, ByteBuffer> customPayload)
                                          throws RequestExecutionException, RequestValidationException
{
    if (customPayload != null)
        requestPayload = customPayload;
    ResultMessage result = QueryProcessor.instance.processBatch(statement, state, options, customPayload);
    if (customPayload != null)
    {
        result.setCustomPayload(responsePayload);
        responsePayload = null;
    }
    return result;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage execute(QueryState state, QueryOptions options) throws InvalidRequestException, TruncateException
{
    try
    {
        CFMetaData metaData = Schema.instance.getCFMetaData(keyspace(), columnFamily());
        if (metaData.isView())
            throw new InvalidRequestException("Cannot TRUNCATE materialized view directly; must truncate base table instead");

        StorageProxy.truncateBlocking(keyspace(), columnFamily());
    }
    catch (UnavailableException | TimeoutException | IOException e)
    {
        throw new TruncateException(e);
    }
    return null;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public static ResultMessage processPrepared(CQLStatement statement, QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
    List<ByteBuffer> variables = options.getValues();
    // Check to see if there are any bound variables to verify
    if (!(variables.isEmpty() && (statement.getBoundsTerms() == 0)))
    {
        if (variables.size() != statement.getBoundsTerms())
            throw new InvalidRequestException(String.format("there were %d markers(?) in CQL but %d bound variables",
                                                            statement.getBoundsTerms(),
                                                            variables.size()));

        // at this point there is a match in count between markers and variables that is non-zero

        if (logger.isTraceEnabled())
            for (int i = 0; i < variables.size(); i++)
                logger.trace("[{}] '{}'", i+1, variables.get(i));
    }

    return processStatement(statement, queryState, options);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage.Rows executeInternal(QueryState state) throws RequestExecutionException, RequestValidationException
{
    List<ByteBuffer> variables = Collections.emptyList();
    int limit = getLimit(variables);
    long now = System.currentTimeMillis();
    List<Row> rows;
    if (isKeyRange || usesSecondaryIndexing)
    {
        RangeSliceCommand command = getRangeCommand(variables, limit, now);
        rows = command == null ? Collections.<Row>emptyList() : command.executeLocally();
    }
    else
    {
        List<ReadCommand> commands = getSliceCommands(variables, limit, now);
        rows = commands == null ? Collections.<Row>emptyList() : readLocally(keyspace(), commands);
    }

    return processResults(rows, variables, limit, now);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage.Rows executeInternal(QueryState state, QueryOptions options, int nowInSec) throws RequestExecutionException, RequestValidationException
{
    int userLimit = getLimit(options);
    ReadQuery query = getQuery(options, nowInSec, userLimit);
    int pageSize = getPageSize(options);

    try (ReadOrderGroup orderGroup = query.startOrderGroup())
    {
        if (pageSize <= 0 || query.limits().count() <= pageSize)
        {
            try (PartitionIterator data = query.executeInternal(orderGroup))
            {
                return processResults(data, options, nowInSec, userLimit);
            }
        }
        else
        {
            QueryPager pager = query.getPager(options.getPagingState(), options.getProtocolVersion());
            return execute(Pager.forInternalQuery(pager, orderGroup), options, pageSize, nowInSec, userLimit);
        }
    }
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage processPrepared(CQLStatement statement,
                                     QueryState state,
                                     QueryOptions options,
                                     Map<String, ByteBuffer> customPayload)
                                             throws RequestExecutionException, RequestValidationException
{
    if (customPayload != null)
        requestPayload = customPayload;
    ResultMessage result = QueryProcessor.instance.processPrepared(statement, state, options, customPayload);
    if (customPayload != null)
    {
        result.setCustomPayload(responsePayload);
        responsePayload = null;
    }
    return result;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
private Set<String> executeCQL(String rawStatement) throws Exception
{
    SelectStatement statement = (SelectStatement) QueryProcessor.parseStatement(rawStatement).prepare().statement;
    ResultMessage.Rows cqlRows = statement.executeInternal(QueryState.forInternalCalls(), new QueryOptions(ConsistencyLevel.LOCAL_ONE, Collections.<ByteBuffer>emptyList()));

    Set<String> results = new TreeSet<>();
    for (CqlRow row : cqlRows.toThriftResult().getRows())
    {
        for (org.apache.cassandra.thrift.Column col : row.columns)
        {
            String columnName = UTF8Type.instance.getString(col.bufferForName());
            if (columnName.equals("key"))
                results.add(AsciiType.instance.getString(col.bufferForValue()));
        }
    }

    return results;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
InternalStateInstance()
{
    ClientState state = ClientState.forInternalCalls();
    try
    {
        state.setKeyspace(Keyspace.SYSTEM_KS);
    }
    catch (InvalidRequestException e)
    {
        throw new RuntimeException();
    }
    this.queryState = new QueryState(state);
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage processStatement(CQLStatement statement, QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
    logger.trace("Process {} @CL.{}", statement, options.getConsistency());
    ClientState clientState = queryState.getClientState();
    statement.checkAccess(clientState);
    statement.validate(clientState);

    ResultMessage result = statement.execute(queryState, options);
    return result == null ? new ResultMessage.Void() : result;
}
 

import org.apache.cassandra.service.QueryState; //导入依赖的package包/类
public ResultMessage.Prepared prepare(String query,
                                      QueryState state,
                                      Map<String, ByteBuffer> customPayload)
                                              throws RequestValidationException
{
    if (customPayload != null)
        requestPayload = customPayload;
    ResultMessage.Prepared result = QueryProcessor.instance.prepare(query, state, customPayload);
    if (customPayload != null)
    {
        result.setCustomPayload(responsePayload);
        responsePayload = null;
    }
    return result;
}