import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
private void assertUnauthorized(BatchStatement batch, Iterable<String> functionNames) throws Throwable
{
    try
    {
        batch.checkAccess(clientState);
        fail("Expected an UnauthorizedException, but none was thrown");
    }
    catch (UnauthorizedException e)
    {
        String functions = String.format("(%s)", Joiner.on("|").join(functionNames));
        assertTrue(e.getLocalizedMessage()
                    .matches(String.format("User %s has no EXECUTE permission on <function %s\\(\\)> or any of its parents",
                                           roleName,
                                           functions)));
    }
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的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.cql3.statements.BatchStatement; //导入依赖的package包/类
private BatchStatement.Type toType(byte b)
{
    if (b == 0)
        return BatchStatement.Type.LOGGED;
    else if (b == 1)
        return BatchStatement.Type.UNLOGGED;
    else if (b == 2)
        return BatchStatement.Type.COUNTER;
    else
        throw new ProtocolException("Invalid BATCH message type " + b);
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
private byte fromType(BatchStatement.Type type)
{
    switch (type)
    {
        case LOGGED:   return 0;
        case UNLOGGED: return 1;
        case COUNTER:  return 2;
        default:
            throw new AssertionError();
    }
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public BatchMessage(BatchStatement.Type type, List<Object> queryOrIdList, List<List<ByteBuffer>> values, QueryOptions options)
{
    super(Message.Type.BATCH);
    this.type = type;
    this.queryOrIdList = queryOrIdList;
    this.values = values;
    this.options = options;
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public BatchMessage(BatchStatement.Type type, List<Object> queryOrIdList, List<List<ByteBuffer>> values, ConsistencyLevel consistency)
{
    super(Message.Type.BATCH);
    this.type = type;
    this.queryOrIdList = queryOrIdList;
    this.values = values;
    this.consistency = consistency;
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public ResultMessage processBatch(BatchStatement statement,
                                  QueryState state,
                                  BatchQueryOptions options,
                                  Map<String, ByteBuffer> customPayload)
{
    ResultMessage result = queryProcessor.processBatch(statement, state, options, customPayload);
    result.setCustomPayload(customPayload);
    return result;
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
private void executeLoggedBatch(List<CQLStatement> statements)
throws RequestExecutionException, RequestValidationException
{
    BatchStatement batch = new BatchStatement(0,
                                              BatchStatement.Type.LOGGED,
                                              Lists.newArrayList(Iterables.filter(statements, ModificationStatement.class)),
                                              Attributes.none());
    QueryProcessor.instance.processBatch(batch,
                                         QueryState.forInternalCalls(),
                                         BatchQueryOptions.withoutPerStatementVariables(QueryOptions.DEFAULT));

}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public BatchMessage(BatchStatement.Type type, List<Object> queryOrIdList, List<List<ByteBuffer>> values, QueryOptions options)
{
    super(Message.Type.BATCH);
    this.batchType = type;
    this.queryOrIdList = queryOrIdList;
    this.values = values;
    this.options = options;
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
@Test
public void testBatchStatement() throws Throwable
{
    String iFunc2 = createEchoFunction("int");
    List<ModificationStatement> statements = new ArrayList<>();
    statements.add(modificationStatement(cql("INSERT INTO %s (key, i_cc, t_cc) VALUES (%s, 0, 'foo')",
                                             functionCall(iFunc, "0"))));
    statements.add(modificationStatement(cql("INSERT INTO %s (key, i_cc, t_cc) VALUES (1, %s, 'foo')",
                                             functionCall(iFunc2, "1"))));
    statements.add(modificationStatement(cql("INSERT INTO %s (key, i_cc, t_cc) VALUES (2, 2, %s)",
                                             functionCall(tFunc, "'foo'"))));

    BatchStatement batch = new BatchStatement(-1, BatchStatement.Type.LOGGED, statements, Attributes.none());
    assertFunctions(batch, iFunc, iFunc2, tFunc);
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
@Test
public void testBatchStatementWithConditions() throws Throwable
{
    List<ModificationStatement> statements = new ArrayList<>();
    statements.add(modificationStatement(cql("UPDATE %s SET i_val = %s WHERE key=0 AND i_cc=0 and t_cc='foo' IF l_val = %s",
                                             functionCall(iFunc, "0"), functionCall(lFunc, "[1]"))));
    statements.add(modificationStatement(cql("UPDATE %s SET i_val = %s WHERE key=0 AND i_cc=1 and t_cc='foo' IF s_val = %s",
                                             functionCall(iFunc, "0"), functionCall(sFunc, "{1}"))));

    BatchStatement batch = new BatchStatement(-1, BatchStatement.Type.LOGGED, statements, Attributes.none());
    assertFunctions(batch, iFunc, lFunc, sFunc);
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
@Test
public void testBatchStatement() throws Throwable
{
    List<ModificationStatement> statements = new ArrayList<>();
    List<String> functions = new ArrayList<>();
    for (int i = 0; i < 3; i++)
    {
        String functionName = createSimpleFunction();
        ModificationStatement stmt =
        (ModificationStatement) getStatement(String.format("INSERT INTO %s (k, v1, v2) " +
                                                           "VALUES (%s, %s, %s)",
                                                           KEYSPACE + "." + currentTable(),
                                                           i, i, functionCall(functionName)));
        functions.add(functionName);
        statements.add(stmt);
    }
    BatchStatement batch = new BatchStatement(-1, BatchStatement.Type.LOGGED, statements, Attributes.none());
    assertUnauthorized(batch, functions);

    grantExecuteOnFunction(functions.get(0));
    assertUnauthorized(batch, functions.subList(1, functions.size()));

    grantExecuteOnFunction(functions.get(1));
    assertUnauthorized(batch, functions.subList(2, functions.size()));

    grantExecuteOnFunction(functions.get(2));
    batch.checkAccess(clientState);
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public Message.Response execute(QueryState state)
{
    try
    {
        UUID tracingId = null;
        if (isTracingRequested())
        {
            tracingId = UUIDGen.getTimeUUID();
            state.prepareTracingSession(tracingId);
        }

        if (state.traceNextQuery())
        {
            state.createTracingSession();
            // TODO we don't have [typed] access to CQL bind variables here.  CASSANDRA-4560 is open to add support.
            Tracing.instance.begin("Execute batch of CQL3 queries", Collections.<String, String>emptyMap());
        }

        List<ModificationStatement> statements = new ArrayList<ModificationStatement>(queryOrIdList.size());
        for (int i = 0; i < queryOrIdList.size(); i++)
        {
            Object query = queryOrIdList.get(i);
            CQLStatement statement;
            if (query instanceof String)
            {
                statement = QueryProcessor.parseStatement((String)query, state);
            }
            else
            {
                statement = QueryProcessor.getPrepared((MD5Digest)query);
                if (statement == null)
                    throw new PreparedQueryNotFoundException((MD5Digest)query);
            }

            List<ByteBuffer> queryValues = values.get(i);
            if (queryValues.size() != statement.getBoundsTerms())
                throw new InvalidRequestException(String.format("There were %d markers(?) in CQL but %d bound variables",
                                                                statement.getBoundsTerms(),
                                                                queryValues.size()));
            if (!(statement instanceof ModificationStatement))
                throw new InvalidRequestException("Invalid statement in batch: only UPDATE, INSERT and DELETE statements are allowed.");

            ModificationStatement mst = (ModificationStatement)statement;
            if (mst.isCounter())
            {
                if (type != BatchStatement.Type.COUNTER)
                    throw new InvalidRequestException("Cannot include counter statement in a non-counter batch");
            }
            else
            {
                if (type == BatchStatement.Type.COUNTER)
                    throw new InvalidRequestException("Cannot include non-counter statement in a counter batch");
            }
            statements.add(mst);
        }

        // Note: It's ok at this point to pass a bogus value for the number of bound terms in the BatchState ctor
        // (and no value would be really correct, so we prefer passing a clearly wrong one).
        BatchStatement batch = new BatchStatement(-1, type, statements, Attributes.none());
        Message.Response response = QueryProcessor.processBatch(batch, consistency, state, values);

        if (tracingId != null)
            response.setTracingId(tracingId);

        return response;
    }
    catch (Exception e)
    {
        return ErrorMessage.fromException(e);
    }
    finally
    {
        Tracing.instance.stopSession();
    }
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
ResultMessage processBatch(BatchStatement statement,
QueryState state,
BatchQueryOptions options,
Map<String, ByteBuffer> customPayload) throws RequestExecutionException, RequestValidationException;
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
@Test
public void testMessagePayload() throws Throwable
{
    QueryHandler queryHandler = (QueryHandler) cqlQueryHandlerField.get(null);
    cqlQueryHandlerField.set(null, new TestQueryHandler());
    try
    {
        requireNetwork();

        Assert.assertSame(TestQueryHandler.class, ClientState.getCQLQueryHandler().getClass());

        SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort);
        try
        {
            client.connect(false);

            Map<String, ByteBuffer> reqMap;
            Map<String, ByteBuffer> respMap;

            QueryMessage queryMessage = new QueryMessage(
                                                        "CREATE TABLE " + KEYSPACE + ".atable (pk int PRIMARY KEY, v text)",
                                                        QueryOptions.DEFAULT
            );
            PrepareMessage prepareMessage = new PrepareMessage("SELECT * FROM " + KEYSPACE + ".atable");

            reqMap = Collections.singletonMap("foo", bytes(42));
            responsePayload = respMap = Collections.singletonMap("bar", bytes(42));
            queryMessage.setCustomPayload(reqMap);
            Message.Response queryResponse = client.execute(queryMessage);
            payloadEquals(reqMap, requestPayload);
            payloadEquals(respMap, queryResponse.getCustomPayload());

            reqMap = Collections.singletonMap("foo", bytes(43));
            responsePayload = respMap = Collections.singletonMap("bar", bytes(43));
            prepareMessage.setCustomPayload(reqMap);
            ResultMessage.Prepared prepareResponse = (ResultMessage.Prepared) client.execute(prepareMessage);
            payloadEquals(reqMap, requestPayload);
            payloadEquals(respMap, prepareResponse.getCustomPayload());

            ExecuteMessage executeMessage = new ExecuteMessage(prepareResponse.statementId, QueryOptions.DEFAULT);
            reqMap = Collections.singletonMap("foo", bytes(44));
            responsePayload = respMap = Collections.singletonMap("bar", bytes(44));
            executeMessage.setCustomPayload(reqMap);
            Message.Response executeResponse = client.execute(executeMessage);
            payloadEquals(reqMap, requestPayload);
            payloadEquals(respMap, executeResponse.getCustomPayload());

            BatchMessage batchMessage = new BatchMessage(BatchStatement.Type.UNLOGGED,
                                                         Collections.<Object>singletonList("INSERT INTO " + KEYSPACE + ".atable (pk,v) VALUES (1, 'foo')"),
                                                         Collections.singletonList(Collections.<ByteBuffer>emptyList()),
                                                         QueryOptions.DEFAULT);
            reqMap = Collections.singletonMap("foo", bytes(45));
            responsePayload = respMap = Collections.singletonMap("bar", bytes(45));
            batchMessage.setCustomPayload(reqMap);
            Message.Response batchResponse = client.execute(batchMessage);
            payloadEquals(reqMap, requestPayload);
            payloadEquals(respMap, batchResponse.getCustomPayload());
        }
        finally
        {
            client.close();
        }
    }
    finally
    {
        cqlQueryHandlerField.set(null, queryHandler);
    }
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public Message.Response execute(QueryState state)
{
    try
    {
        UUID tracingId = null;
        if (isTracingRequested())
        {
            tracingId = UUIDGen.getTimeUUID();
            state.prepareTracingSession(tracingId);
        }

        if (state.traceNextQuery())
        {
            state.createTracingSession();
            // TODO we don't have [typed] access to CQL bind variables here.  CASSANDRA-4560 is open to add support.
            Tracing.instance.begin("Execute batch of CQL3 queries", Collections.<String, String>emptyMap());
        }

        List<ModificationStatement> statements = new ArrayList<ModificationStatement>(queryOrIdList.size());
        for (int i = 0; i < queryOrIdList.size(); i++)
        {
            Object query = queryOrIdList.get(i);
            CQLStatement statement;
            if (query instanceof String)
            {
                statement = QueryProcessor.parseStatement((String)query, state);
            }
            else
            {
                statement = QueryProcessor.getPrepared((MD5Digest)query);
                if (statement == null)
                    throw new PreparedQueryNotFoundException((MD5Digest)query);
            }

            List<ByteBuffer> queryValues = values.get(i);
            if (queryValues.size() != statement.getBoundTerms())
                throw new InvalidRequestException(String.format("There were %d markers(?) in CQL but %d bound variables",
                                                                statement.getBoundTerms(),
                                                                queryValues.size()));
            if (!(statement instanceof ModificationStatement))
                throw new InvalidRequestException("Invalid statement in batch: only UPDATE, INSERT and DELETE statements are allowed.");

            ModificationStatement mst = (ModificationStatement)statement;
            if (mst.isCounter())
            {
                if (type != BatchStatement.Type.COUNTER)
                    throw new InvalidRequestException("Cannot include counter statement in a non-counter batch");
            }
            else
            {
                if (type == BatchStatement.Type.COUNTER)
                    throw new InvalidRequestException("Cannot include non-counter statement in a counter batch");
            }
            statements.add(mst);
        }

        // Note: It's ok at this point to pass a bogus value for the number of bound terms in the BatchState ctor
        // (and no value would be really correct, so we prefer passing a clearly wrong one).
        BatchStatement batch = new BatchStatement(-1, type, statements, Attributes.none());
        Message.Response response = QueryProcessor.processBatch(batch, consistency, state, values, queryOrIdList);

        if (tracingId != null)
            response.setTracingId(tracingId);

        return response;
    }
    catch (Exception e)
    {
        return ErrorMessage.fromException(e);
    }
    finally
    {
        Tracing.instance.stopSession();
    }
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public Message.Response execute(QueryState state)
{
    try
    {
        UUID tracingId = null;
        if (isTracingRequested())
        {
            tracingId = UUIDGen.getTimeUUID();
            state.prepareTracingSession(tracingId);
        }

        if (state.traceNextQuery())
        {
            state.createTracingSession();
            // TODO we don't have [typed] access to CQL bind variables here.  CASSANDRA-4560 is open to add support.
            Tracing.instance.begin("Execute batch of CQL3 queries", Collections.<String, String>emptyMap());
        }

        QueryHandler handler = state.getClientState().getCQLQueryHandler();
        List<ModificationStatement> statements = new ArrayList<ModificationStatement>(queryOrIdList.size());
        for (int i = 0; i < queryOrIdList.size(); i++)
        {
            Object query = queryOrIdList.get(i);
            CQLStatement statement;
            if (query instanceof String)
            {
                statement = QueryProcessor.parseStatement((String)query, state);
            }
            else
            {
                statement = handler.getPrepared((MD5Digest)query);
                if (statement == null)
                    throw new PreparedQueryNotFoundException((MD5Digest)query);
            }

            List<ByteBuffer> queryValues = values.get(i);
            if (queryValues.size() != statement.getBoundTerms())
                throw new InvalidRequestException(String.format("There were %d markers(?) in CQL but %d bound variables",
                                                                statement.getBoundTerms(),
                                                                queryValues.size()));
            if (!(statement instanceof ModificationStatement))
                throw new InvalidRequestException("Invalid statement in batch: only UPDATE, INSERT and DELETE statements are allowed.");

            ModificationStatement mst = (ModificationStatement)statement;
            if (mst.isCounter())
            {
                if (type != BatchStatement.Type.COUNTER)
                    throw new InvalidRequestException("Cannot include counter statement in a non-counter batch");
            }
            else
            {
                if (type == BatchStatement.Type.COUNTER)
                    throw new InvalidRequestException("Cannot include non-counter statement in a counter batch");
            }
            statements.add(mst);
        }

        // Note: It's ok at this point to pass a bogus value for the number of bound terms in the BatchState ctor
        // (and no value would be really correct, so we prefer passing a clearly wrong one).
        BatchStatement batch = new BatchStatement(-1, type, statements, Attributes.none());
        Message.Response response = handler.processBatch(batch, state, new BatchQueryOptions(consistency, values, queryOrIdList));

        if (tracingId != null)
            response.setTracingId(tracingId);

        return response;
    }
    catch (Exception e)
    {
        return ErrorMessage.fromException(e);
    }
    finally
    {
        Tracing.instance.stopSession();
    }
}
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
/**
 * Perform post-processing on a CQL batch directly after
 * it being executed by the QueryProcessor.
 *
 * @param batch the CQL batch to perform post-processing on
 * @param context execution context containing additional info
 *                about the operation and batch
 * @throws RequestExecutionException, RequestValidationException
 */
void processBatch(BatchStatement batch, BatchExecutionContext context) throws RequestExecutionException, RequestValidationException;
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
/**
 * Perform pre-processing on a CQL batch prior to it being
 * executed by the QueryProcessor. If required, implementations
 * may modify the batch & its component statements as the returned
 * instance is what is actually executed.
 *
 * @param batch the CQL batch to perform pre-processing on
 * @param context execution context containing additional info
 *                about the operation and batch
 * @return the actual batch that will be executed, potentially
 *         a modification of the initial batch
 * @throws RequestExecutionException, RequestValidationException
 */
BatchStatement processBatch(BatchStatement batch, BatchExecutionContext context) throws RequestExecutionException, RequestValidationException;
 

import org.apache.cassandra.cql3.statements.BatchStatement; //导入依赖的package包/类
public ResultMessage processBatch(BatchStatement statement, QueryState state, BatchQueryOptions options) throws RequestExecutionException, RequestValidationException;