import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
private void addStatementMutations(ModificationStatement statement,
                                   List<ByteBuffer> variables,
                                   boolean local,
                                   ConsistencyLevel cl,
                                   long now,
                                   Map<Pair<String, ByteBuffer>, IMutation> mutations)
throws RequestExecutionException, RequestValidationException
{
    // Group mutation together, otherwise they won't get applied atomically
    for (IMutation m : statement.getMutations(variables, local, cl, attrs.getTimestamp(now, variables), true))
    {
        Pair<String, ByteBuffer> key = Pair.create(m.getKeyspaceName(), m.key());
        IMutation existing = mutations.get(key);

        if (existing == null)
        {
            mutations.put(key, m);
        }
        else
        {
            existing.addAll(m);
        }
    }
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public List<IMutation> prepareRowMutations(String keyspace, ThriftClientState clientState, Long timestamp, List<ByteBuffer> variables)
throws InvalidRequestException, UnauthorizedException
{
    CFMetaData metadata = validateColumnFamily(keyspace, columnFamily);

    clientState.hasColumnFamilyAccess(keyspace, columnFamily, Permission.MODIFY);
    AbstractType<?> keyType = Schema.instance.getCFMetaData(keyspace, columnFamily).getKeyValidator();

    List<IMutation> rowMutations = new ArrayList<IMutation>(keys.size());

    for (Term key : keys)
    {
        rowMutations.add(mutationForKey(key.getByteBuffer(keyType, variables), keyspace, timestamp, clientState, variables, metadata));
    }

    return rowMutations;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public Collection<RowMutation> execute(Collection<? extends IMutation> updates) throws InvalidRequestException
{
    boolean hasCounters = false;
    Collection<RowMutation> tmutations = null;
    for (IMutation mutation : updates)
    {
        for (ColumnFamily cf : mutation.getColumnFamilies())
        {
            List<RowMutation> intermediate = execute(mutation.key(), cf);
            if (intermediate == null)
                continue;

            validate(intermediate);
            if (tmutations == null)
                tmutations = intermediate;
            else
                tmutations.addAll(intermediate);
        }
        if (mutation instanceof CounterMutation)
            hasCounters = true;
    }
    if (tmutations != null && hasCounters)
        throw new InvalidRequestException("Counter mutations and trigger mutations cannot be applied together atomically.");
    return tmutations;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public List<IMutation> getMutations(String keyspace, ThriftClientState clientState, List<ByteBuffer> variables)
throws InvalidRequestException, UnauthorizedException
{
    List<IMutation> batch = new LinkedList<IMutation>();

    for (AbstractModification statement : statements) {
        batch.addAll(statement.prepareRowMutations(keyspace, clientState, timestamp, variables));
    }

    return batch;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public List<IMutation> prepareRowMutations(String keyspace, ThriftClientState clientState, Long timestamp, List<ByteBuffer> variables)
throws InvalidRequestException, UnauthorizedException
{
    CFMetaData metadata = validateColumnFamily(keyspace, columnFamily);

    clientState.hasColumnFamilyAccess(keyspace, columnFamily, Permission.MODIFY);
    AbstractType<?> keyType = Schema.instance.getCFMetaData(keyspace, columnFamily).getKeyValidator();

    List<IMutation> mutations = new ArrayList<IMutation>(keys.size());

    for (Term key : keys)
        mutations.add(mutationForKey(key.getByteBuffer(keyType, variables), keyspace, timestamp, clientState, variables, metadata));

    return mutations;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public static IMutation processWriteRequest(IMutation mutation)
{
    if (requestSinks.isEmpty())
        return mutation;

    for (IRequestSink rs : requestSinks)
    {
        mutation = rs.handleWriteRequest(mutation);
        if (mutation == null)
            return null;
    }
    return mutation;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
private Collection<? extends IMutation> getMutations(List<ByteBuffer> variables, boolean local, ConsistencyLevel cl, long now)
throws RequestExecutionException, RequestValidationException
{
    Map<Pair<String, ByteBuffer>, IMutation> mutations = new HashMap<Pair<String, ByteBuffer>, IMutation>();
    for (ModificationStatement statement : statements)
        addStatementMutations(statement, variables, local, cl, now, mutations);

    return mutations.values();
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
/** {@inheritDoc} */
public List<IMutation> prepareRowMutations(String keyspace, ThriftClientState clientState, Long timestamp, List<ByteBuffer> variables)
throws InvalidRequestException, UnauthorizedException
{
    boolean hasCommutativeOperation = false;

    for (Map.Entry<Term, Operation> column : getColumns().entrySet())
    {
        if (!column.getValue().isUnary())
            hasCommutativeOperation = true;

        if (hasCommutativeOperation && column.getValue().isUnary())
            throw new InvalidRequestException("Mix of commutative and non-commutative operations is not allowed.");
    }

    CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, hasCommutativeOperation);
    if (hasCommutativeOperation)
        getConsistencyLevel().validateCounterForWrite(metadata);

    QueryProcessor.validateKeyAlias(metadata, keyName);

    clientState.hasColumnFamilyAccess(keyspace, columnFamily, Permission.MODIFY);

    List<IMutation> rowMutations = new LinkedList<IMutation>();

    for (Term key: keys)
    {
        rowMutations.add(mutationForKey(keyspace, key.getByteBuffer(getKeyType(keyspace),variables), metadata, timestamp, clientState, variables));
    }

    return rowMutations;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public Collection<? extends IMutation> getMutations(List<ByteBuffer> variables, boolean local, ConsistencyLevel cl, long now)
throws RequestExecutionException, RequestValidationException
{
    Map<Pair<String, ByteBuffer>, IMutation> mutations = new HashMap<Pair<String, ByteBuffer>, IMutation>();
    for (ModificationStatement statement : statements)
    {
        // Group mutation together, otherwise they won't get applied atomically
        for (IMutation m : statement.getMutations(variables, local, cl, getTimestamp(now)))
        {
            if (m instanceof CounterMutation && type != Type.COUNTER)
                throw new InvalidRequestException("Counter mutations are only allowed in COUNTER batches");

            if (m instanceof RowMutation && type == Type.COUNTER)
                throw new InvalidRequestException("Only counter mutations are allowed in COUNTER batches");

            Pair<String, ByteBuffer> key = Pair.create(m.getTable(), m.key());
            IMutation existing = mutations.get(key);

            if (existing == null)
                mutations.put(key, m);
            else
                existing.addAll(m);
        }
    }

    return mutations.values();
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
/** {@inheritDoc} */
public List<IMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
{
    List<String> cfamsSeen = new ArrayList<String>();

    boolean hasCommutativeOperation = false;

    for (Map.Entry<Term, Operation> column : getColumns().entrySet())
    {
        if (!column.getValue().isUnary())
            hasCommutativeOperation = true;

        if (hasCommutativeOperation && column.getValue().isUnary())
            throw new InvalidRequestException("Mix of commutative and non-commutative operations is not allowed.");
    }

    CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, hasCommutativeOperation);
    if (hasCommutativeOperation)
        validateCommutativeForWrite(metadata, cLevel);

    QueryProcessor.validateKeyAlias(metadata, keyName);

    // Avoid unnecessary authorizations.
    if (!(cfamsSeen.contains(columnFamily)))
    {
        clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
        cfamsSeen.add(columnFamily);
    }

    List<IMutation> rowMutations = new LinkedList<IMutation>();

    for (Term key: keys)
    {
        rowMutations.add(mutationForKey(keyspace, key.getByteBuffer(getKeyType(keyspace)), metadata, timestamp));
    }

    return rowMutations;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public List<IMutation> getMutations(String keyspace, ClientState clientState) throws InvalidRequestException
{
    List<IMutation> batch = new LinkedList<IMutation>();

    for (AbstractModification statement : statements) {
        batch.addAll(statement.prepareRowMutations(keyspace, clientState, timestamp));
    }

    return batch;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
/** {@inheritDoc} */
public List<IMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
{
    clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE);
    AbstractType<?> keyType = DatabaseDescriptor.getCFMetaData(keyspace, columnFamily).getKeyValidator();

    List<IMutation> rowMutations = new ArrayList<IMutation>();

    for (Term key : keys)
    {
        rowMutations.add(mutationForKey(key.getByteBuffer(keyType), keyspace, timestamp));
    }

    return rowMutations;
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
private void execute(Collection<? extends IMutation> mutations, ConsistencyLevel cl) throws RequestExecutionException, RequestValidationException
{
    boolean mutateAtomic = (type == Type.LOGGED && mutations.size() > 1);
    StorageProxy.mutateWithTriggers(mutations, cl, mutateAtomic);
}
 

import org.apache.cassandra.db.IMutation; //导入依赖的package包/类
public ResultMessage executeInternal(QueryState queryState) throws RequestValidationException, RequestExecutionException
{
    for (IMutation mutation : getMutations(Collections.<ByteBuffer>emptyList(), true, null, queryState.getTimestamp()))
        mutation.apply();
    return null;
}