cassandra/src/java/org/apache/cassandra/db/Mutation.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.cassandra.db;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.function.Predicate;
import java.util.function.Supplier;

import javax.annotation.Nullable;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableMap;

import org.apache.commons.lang3.StringUtils;

import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ReadCommand.PotentialTxnConflicts;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.TeeDataInputPlus;
import org.apache.cassandra.locator.ReplicaPlan;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.AbstractWriteResponseHandler;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.concurrent.Future;

import static com.google.common.base.Preconditions.checkState;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.net.MessagingService.VERSION_50;
import static org.apache.cassandra.net.MessagingService.VERSION_60;
import static org.apache.cassandra.utils.MonotonicClock.Global.approxTime;

public class Mutation implements IMutation, Supplier<Mutation>
{
    public static final MutationSerializer serializer = new MutationSerializer();
    public static final int ALLOW_POTENTIAL_TRANSACTION_CONFLICTS = 0x01;


    // todo this is redundant
    // when we remove it, also restore SerializationsTest.testMutationRead to not regenerate new Mutations each test
    private final String keyspaceName;

    private final DecoratedKey key;
    // map of column family id to mutations for that column family.
    private final ImmutableMap<TableId, PartitionUpdate> modifications;

    // Time at which this mutation or the builder that built it was instantiated
    final long approxCreatedAtNanos;
    // keep track of when mutation has started waiting for a MV partition lock

    final static AtomicLongFieldUpdater<Mutation> viewLockAcquireStartUpdater =
        AtomicLongFieldUpdater.newUpdater(Mutation.class, "viewLockAcquireStart");
    volatile long viewLockAcquireStart;

    private final boolean cdcEnabled;

    private static final int SERIALIZATION_VERSION_COUNT = MessagingService.Version.values().length;
    // Contains serialized representations of this mutation.
    // Note: The cached serializations can be cleared when CoordinatorBehindException is being retried
    private final Serialization[] cachedSerializations = new Serialization[SERIALIZATION_VERSION_COUNT];

    /** @see CassandraRelevantProperties#CACHEABLE_MUTATION_SIZE_LIMIT */
    private static final long CACHEABLE_MUTATION_SIZE_LIMIT = CassandraRelevantProperties.CACHEABLE_MUTATION_SIZE_LIMIT.getLong();

    // Paxos & Accord manage conflicts directly and needs to apply mutations to tables/ranges
    // that are only safe to write to from a transaction system.
    // Don't refuse to apply this mutation because it should go through a transaction system
    // because it is being applied by one or in a context where transaction conflicts don't occur
    private PotentialTxnConflicts potentialTxnConflicts;

    public Mutation(PartitionUpdate update)
    {
        this(update, PotentialTxnConflicts.DISALLOW);
    }

    public Mutation(PartitionUpdate update, PotentialTxnConflicts potentialTxnConflicts)
    {
        this(update.metadata().keyspace, update.partitionKey(), ImmutableMap.of(update.metadata().id, update), approxTime.now(), update.metadata().params.cdc, potentialTxnConflicts);
    }

    public Mutation(String keyspaceName, DecoratedKey key, ImmutableMap<TableId, PartitionUpdate> modifications, long approxCreatedAtNanos, PotentialTxnConflicts potentialTxnConflicts)
    {
        this(keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled(modifications.values()), potentialTxnConflicts);
    }

    public Mutation(String keyspaceName, DecoratedKey key, ImmutableMap<TableId, PartitionUpdate> modifications, long approxCreatedAtNanos, boolean cdcEnabled, PotentialTxnConflicts potentialTxnConflicts)
    {
        this.keyspaceName = keyspaceName;
        this.key = key;
        this.modifications = modifications;
        this.cdcEnabled = cdcEnabled;
        this.approxCreatedAtNanos = approxCreatedAtNanos;
        this.potentialTxnConflicts = potentialTxnConflicts;
    }

    private static boolean cdcEnabled(Iterable<PartitionUpdate> modifications)
    {
        boolean cdc = false;
        for (PartitionUpdate pu : modifications)
            cdc |= pu.metadata().params.cdc;
        return cdc;
    }

    @Override
    public @Nullable Mutation filter(Predicate<TableId> predicate)
    {
        boolean allMatch = true;
        boolean noneMatch = true;
        for (TableId tableId : modifications.keySet())
        {
            boolean test = predicate.test(tableId);
            allMatch &= test;
            noneMatch &= !test;
        }
        if (allMatch)
            return this;
        if (noneMatch)
            return null;

        ImmutableMap.Builder<TableId, PartitionUpdate> builder = new ImmutableMap.Builder<>();
        for (Map.Entry<TableId, PartitionUpdate> update : modifications.entrySet())
            if (predicate.test(update.getKey()))
                builder.put(update);

        Map<TableId, PartitionUpdate> updates = builder.build();
        checkState(!updates.isEmpty(), "Updates should not be empty");
        return new Mutation(keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts);
    }

    public @Nullable Mutation without(TableId tableId)
    {
        return filter(otherTableId -> !tableId.equals(otherTableId));
    }

    public String getKeyspaceName()
    {
        return keyspaceName;
    }

    public Collection<TableId> getTableIds()
    {
        return modifications.keySet();
    }

    public DecoratedKey key()
    {
        return key;
    }

    public ImmutableCollection<PartitionUpdate> getPartitionUpdates()
    {
        return modifications.values();
    }

    public long getApproxCreatedAtNanos()
    {
        return approxCreatedAtNanos;
    }

    @Override
    public boolean hasUpdateForTable(TableId tableId)
    {
        return modifications.containsKey(tableId);
    }

    @Override
    public Supplier<Mutation> hintOnFailure()
    {
        return this;
    }

    @Override
    public Mutation get()
    {
        return this;
    }

    public void validateSize(int version, int overhead)
    {
        long totalSize = serializedSize(version) + overhead;
        if(totalSize > MAX_MUTATION_SIZE)
        {
            CommitLog.instance.metrics.oversizedMutations.mark();
            throw new MutationExceededMaxSizeException(this, version, totalSize);
        }
    }

    public PartitionUpdate getPartitionUpdate(TableMetadata table)
    {
        return table == null ? null : modifications.get(table.id);
    }

    public boolean isEmpty()
    {
        return modifications.isEmpty();
    }

    /**
     * Creates a new mutation that merges all the provided mutations.
     *
     * @param mutations the mutations to merge together. All mutation must be
     * on the same keyspace and partition key. There should also be at least one
     * mutation.
     * @return a mutation that contains all the modifications contained in {@code mutations}.
     *
     * @throws IllegalArgumentException if not all the mutations are on the same
     * keyspace and key.
     */
    public static Mutation merge(Collection<Mutation> mutations)
    {
        assert !mutations.isEmpty();

        if (mutations.size() == 1)
            return mutations.iterator().next();

        Set<TableId> updatedTables = new HashSet<>();
        String ks = null;
        DecoratedKey key = null;
        PotentialTxnConflicts potentialTxnConflicts = null;
        for (Mutation mutation : mutations)
        {
            if (potentialTxnConflicts != null && potentialTxnConflicts != mutation.potentialTxnConflicts)
                throw new IllegalArgumentException("Can't merge mutations with differing policies on allowing potential transaction conflicts");
            potentialTxnConflicts = mutation.potentialTxnConflicts;
            updatedTables.addAll(mutation.modifications.keySet());
            if (ks != null && !ks.equals(mutation.keyspaceName))
                throw new IllegalArgumentException();
            if (key != null && !key.equals(mutation.key))
                throw new IllegalArgumentException();
            ks = mutation.keyspaceName;
            key = mutation.key;
        }

        List<PartitionUpdate> updates = new ArrayList<>(mutations.size());
        ImmutableMap.Builder<TableId, PartitionUpdate> modifications = new ImmutableMap.Builder<>();
        for (TableId table : updatedTables)
        {
            for (Mutation mutation : mutations)
            {
                PartitionUpdate upd = mutation.modifications.get(table);
                if (upd != null)
                    updates.add(upd);
            }

            if (updates.isEmpty())
                continue;

            modifications.put(table, updates.size() == 1 ? updates.get(0) : PartitionUpdate.merge(updates));
            updates.clear();
        }
        return new Mutation(ks, key, modifications.build(), approxTime.now(), potentialTxnConflicts);
    }

    public Future<?> applyFuture()
    {
        Keyspace ks = Keyspace.open(keyspaceName);
        return ks.applyFuture(this, Keyspace.open(keyspaceName).getMetadata().params.durableWrites, true);
    }

    private void apply(Keyspace keyspace, boolean durableWrites, boolean isDroppable)
    {
        keyspace.apply(this, durableWrites, true, isDroppable);
    }

    public void apply(boolean durableWrites, boolean isDroppable)
    {
        apply(Keyspace.open(keyspaceName), durableWrites, isDroppable);
    }

    public void apply(boolean durableWrites)
    {
        apply(durableWrites, true);
    }

    /*
     * This is equivalent to calling commit. Applies the changes to
     * to the keyspace that is obtained by calling Keyspace.open().
     */
    public void apply()
    {
        Keyspace keyspace = Keyspace.open(keyspaceName);
        apply(keyspace, keyspace.getMetadata().params.durableWrites, true);
    }

    public void applyUnsafe()
    {
        apply(false);
    }

    public long getTimeout(TimeUnit unit)
    {
        return DatabaseDescriptor.getWriteRpcTimeout(unit);
    }

    public int smallestGCGS()
    {
        int gcgs = Integer.MAX_VALUE;
        for (PartitionUpdate update : getPartitionUpdates())
            gcgs = Math.min(gcgs, update.metadata().params.gcGraceSeconds);
        return gcgs;
    }

    public boolean trackedByCDC()
    {
        return cdcEnabled;
    }

    public void allowPotentialTransactionConflicts()
    {
        potentialTxnConflicts = PotentialTxnConflicts.ALLOW;
        Arrays.fill(cachedSerializations, null);
    }

    @Override
    public PotentialTxnConflicts potentialTxnConflicts()
    {
        return potentialTxnConflicts;
    }

    private static int potentialTxnConflictsFlag(PotentialTxnConflicts potentialTxnConflicts)
    {
        return potentialTxnConflicts.allowed ? ALLOW_POTENTIAL_TRANSACTION_CONFLICTS : 0;
    }

    public static PotentialTxnConflicts potentialTxnConflicts(int flags)
    {
        return (flags & ALLOW_POTENTIAL_TRANSACTION_CONFLICTS) != 0 ? PotentialTxnConflicts.ALLOW : PotentialTxnConflicts.DISALLOW;
    }

    public String toString()
    {
        return toString(false);
    }

    @Override
    public void clearCachedSerializationsForRetry()
    {
        Arrays.fill(cachedSerializations, null);
    }

    public String toString(boolean shallow)
    {
        StringBuilder buff = new StringBuilder("Mutation(");
        buff.append("keyspace='").append(keyspaceName).append('\'');
        buff.append(", key='").append(ByteBufferUtil.bytesToHex(key.getKey())).append('\'');
        buff.append(", modifications=[");
        if (shallow)
        {
            List<String> cfnames = new ArrayList<>(modifications.size());
            for (TableId tableId : modifications.keySet())
            {
                TableMetadata cfm = Schema.instance.getTableMetadata(tableId);
                cfnames.add(cfm == null ? "-dropped-" : cfm.name);
            }
            buff.append(StringUtils.join(cfnames, ", "));
        }
        else
        {
            buff.append("\n  ").append(StringUtils.join(modifications.values(), "\n  ")).append('\n');
        }
        return buff.append("])").toString();
    }

    private int serializedSize40;
    private int serializedSize50;
    private int serializedSize51;

    public int serializedSize(int version)
    {
        switch (version)
        {
            case VERSION_40:
                if (serializedSize40 == 0)
                    serializedSize40 = (int) serializer.serializedSize(this, VERSION_40);
                return serializedSize40;
            case VERSION_50:
                if (serializedSize50 == 0)
                    serializedSize50 = (int) serializer.serializedSize(this, VERSION_50);
                return serializedSize50;
            case VERSION_60:
                if (serializedSize51 == 0)
                    serializedSize51 = (int) serializer.serializedSize(this, VERSION_60);
                return serializedSize51;
            default:
                throw new IllegalStateException("Unknown serialization version: " + version);
        }
    }

    /**
     * Creates a new simple mutuation builder.
     *
     * @param keyspaceName the name of the keyspace this is a mutation for.
     * @param partitionKey the key of partition this if a mutation for.
     * @return a newly created builder.
     */
    public static SimpleBuilder simpleBuilder(String keyspaceName, DecoratedKey partitionKey)
    {
        return new SimpleBuilders.MutationBuilder(keyspaceName, partitionKey);
    }

    /**
     * Interface for building mutations geared towards human.
     * <p>
     * This should generally not be used when performance matters too much, but provides a more convenient interface to
     * build a mutation than using the class constructor when performance is not of the utmost importance.
     */
    public interface SimpleBuilder
    {
        /**
         * Assume any potential transaction conflicts that might occur by applying this mutation are already
         * being handled by the caller
         * @return this builder
         */
        public SimpleBuilder allowPotentialTxnConflicts();

        /**
         * Sets the timestamp to use for the following additions to this builder or any derived (update or row) builder.
         *
         * @param timestamp the timestamp to use for following additions. If that timestamp hasn't been set, the current
         * time in microseconds will be used.
         * @return this builder.
         */
        public SimpleBuilder timestamp(long timestamp);

        /**
         * Sets the ttl to use for the following additions to this builder or any derived (update or row) builder.
         * <p>
         * Note that the for non-compact tables, this method must be called before any column addition for this
         * ttl to be used for the row {@code LivenessInfo}.
         *
         * @param ttl the ttl to use for following additions. If that ttl hasn't been set, no ttl will be used.
         * @return this builder.
         */
        public SimpleBuilder ttl(int ttl);

        /**
         * Adds an update for table identified by the provided metadata and return a builder for that partition.
         *
         * @param metadata the metadata of the table for which to add an update.
         * @return a builder for the partition identified by {@code metadata} (and the partition key for which this is a
         * mutation of).
         */
        public PartitionUpdate.SimpleBuilder update(TableMetadata metadata);

        /**
         * Adds an update for table identified by the provided name and return a builder for that partition.
         *
         * @param tableName the name of the table for which to add an update.
         * @return a builder for the partition identified by {@code metadata} (and the partition key for which this is a
         * mutation of).
         */
        public PartitionUpdate.SimpleBuilder update(String tableName);

        /**
         * Build the mutation represented by this builder.
         *
         * @return the built mutation.
         */
        public Mutation build();
    }

    public static class MutationSerializer implements IVersionedSerializer<Mutation>
    {
        public void serialize(Mutation mutation, DataOutputPlus out, int version) throws IOException
        {
            serialization(mutation, version).serialize(PartitionUpdate.serializer, mutation, out, version);
        }

        /**
         * Called early during request processing to prevent that {@link #serialization(Mutation, int)} is
         * called concurrently.
         * See {@link org.apache.cassandra.service.StorageProxy#sendToHintedReplicas(Mutation, ReplicaPlan.ForWrite, AbstractWriteResponseHandler, String, Stage)}
         */
        @SuppressWarnings("JavadocReference")
        public void prepareSerializedBuffer(Mutation mutation, int version)
        {
            serialization(mutation, version);
        }

        /**
         * Retrieve the cached serialization of this mutation, or compute and cache said serialization if it doesn't
         * exist yet. Note that this method is _not_ synchronized even though it may (and will often) be called
         * concurrently. Concurrent calls are still safe however, the only risk is that the value is not cached yet,
         * multiple concurrent calls may compute it multiple times instead of just once. This is ok as in practice
         * as we make sure this doesn't happen in the hot path by forcing the initial caching in
         * {@link org.apache.cassandra.service.StorageProxy#sendToHintedReplicas(Mutation, ReplicaPlan.ForWrite, AbstractWriteResponseHandler, String, Stage)}
         * via {@link #prepareSerializedBuffer(Mutation)}, which is the only caller that passes
         * {@code isPrepare==true}.
         */
        @SuppressWarnings("JavadocReference")
        private Serialization serialization(Mutation mutation, int version)
        {
            int versionOrdinal = MessagingService.getVersionOrdinal(version);
            // Retrieves the cached version, or build+cache it if it's not cached already.
            Serialization serialization = mutation.cachedSerializations[versionOrdinal];
            if (serialization == null)
            {
                serialization = new SizeOnlyCacheableSerialization();
                long serializedSize = serialization.serializedSize(PartitionUpdate.serializer, mutation, version);

                // Excessively large mutation objects cause GC pressure and huge allocations when serialized.
                // so we only cache serialized mutations when they are below the defined limit.
                if (serializedSize < CACHEABLE_MUTATION_SIZE_LIMIT)
                {
                    try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
                    {
                        serializeInternal(PartitionUpdate.serializer, mutation, dob, version);
                        serialization = new CachedSerialization(dob.unsafeToByteArray());
                    }
                    catch (IOException e)
                    {
                        throw new RuntimeException(e);
                    }
                }
                mutation.cachedSerializations[versionOrdinal] = serialization;
            }

            return serialization;
        }

        static void serializeInternal(PartitionUpdate.PartitionUpdateSerializer serializer,
                                         Mutation mutation,
                                         DataOutputPlus out,
                                         int version) throws IOException
        {
            Map<TableId, PartitionUpdate> modifications = mutation.modifications;

            if (version >= VERSION_60)
            {
                int flags = 0;
                flags |= potentialTxnConflictsFlag(mutation.potentialTxnConflicts);
                out.write(flags);
            }

            /* serialize the modifications in the mutation */
            int size = modifications.size();
            out.writeUnsignedVInt32(size);

            assert size > 0;
            for (PartitionUpdate partitionUpdate : modifications.values())
            {
                serializer.serialize(partitionUpdate, out, version);
            }
        }

        public Mutation deserialize(DataInputPlus in, int version, DeserializationHelper.Flag flag) throws IOException
        {
            Mutation m;
            TeeDataInputPlus teeIn;
            try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
            {
                teeIn = new TeeDataInputPlus(in, dob, CACHEABLE_MUTATION_SIZE_LIMIT);

                PotentialTxnConflicts potentialTxnConflicts = PotentialTxnConflicts.DISALLOW;
                if (version >= VERSION_60)
                {
                    int flags = teeIn.readByte();
                    potentialTxnConflicts = potentialTxnConflicts(flags);
                }
                int size = teeIn.readUnsignedVInt32();
                assert size > 0;

                PartitionUpdate update = PartitionUpdate.serializer.deserialize(teeIn, version, flag);
                if (size == 1)
                {
                    m = new Mutation(update, potentialTxnConflicts);
                }
                else
                {
                    ImmutableMap.Builder<TableId, PartitionUpdate> modifications = new ImmutableMap.Builder<>();
                    DecoratedKey dk = update.partitionKey();

                    modifications.put(update.metadata().id, update);
                    for (int i = 1; i < size; ++i)
                    {
                        update = PartitionUpdate.serializer.deserialize(teeIn, version, flag);
                        modifications.put(update.metadata().id, update);
                    }
                    m = new Mutation(update.metadata().keyspace, dk, modifications.build(), approxTime.now(), potentialTxnConflicts);
                }

                //Only cache serializations that don't hit the limit
                if (!teeIn.isLimitReached())
                    m.cachedSerializations[MessagingService.getVersionOrdinal(version)] = new CachedSerialization(dob.unsafeToByteArray());

                return m;
            }
        }

        public Mutation deserialize(DataInputPlus in, int version) throws IOException
        {
            return deserialize(in, version, DeserializationHelper.Flag.FROM_REMOTE);
        }

        public long serializedSize(Mutation mutation, int version)
        {
            return serialization(mutation, version).serializedSize(PartitionUpdate.serializer, mutation, version);
        }
    }

    /**
     * There are two implementations of this class. One that keeps the serialized representation on-heap for later
     * reuse and one that doesn't. Keeping all sized mutations around may lead to "bad" GC pressure (G1 GC) due to humongous objects.
     * By default serialized mutations up to 2MB are kept on-heap - see {@link org.apache.cassandra.config.CassandraRelevantProperties#CACHEABLE_MUTATION_SIZE_LIMIT}.
     */
    private static abstract class Serialization
    {
        abstract void serialize(PartitionUpdate.PartitionUpdateSerializer serializer, Mutation mutation, DataOutputPlus out, int version) throws IOException;

        abstract long serializedSize(PartitionUpdate.PartitionUpdateSerializer serializer, Mutation mutation, int version);
    }

    /**
     * Represents the cached serialization of a {@link Mutation} as a {@code byte[]}.
     */
    private static final class CachedSerialization extends Serialization
    {
        private final byte[] serialized;

        CachedSerialization(byte[] serialized)
        {
            this.serialized = Preconditions.checkNotNull(serialized);
        }

        @Override
        void serialize(PartitionUpdate.PartitionUpdateSerializer serializer, Mutation mutation, DataOutputPlus out, int version) throws IOException
        {
            out.write(serialized);
        }

        @Override
        long serializedSize(PartitionUpdate.PartitionUpdateSerializer serializer, Mutation mutation, int version)
        {
            return serialized.length;
        }
    }

    /**
     * Represents a non-cacheable serialization of a {@link Mutation}, only the size of the mutation is lazily cached.
     */
    private static final class SizeOnlyCacheableSerialization extends Serialization
    {
        private volatile long size;

        @Override
        void serialize(PartitionUpdate.PartitionUpdateSerializer serializer, Mutation mutation, DataOutputPlus out, int version) throws IOException
        {
            MutationSerializer.serializeInternal(serializer, mutation, out, version);
        }

        @Override
        long serializedSize(PartitionUpdate.PartitionUpdateSerializer serializer, Mutation mutation, int version)
        {
            long size = this.size;
            if (size == 0L)
            {
                if (version >= VERSION_60)
                    size += TypeSizes.sizeof((byte)ALLOW_POTENTIAL_TRANSACTION_CONFLICTS); // flags
                size += TypeSizes.sizeofUnsignedVInt(mutation.modifications.size());
                for (PartitionUpdate partitionUpdate : mutation.modifications.values())
                    size += serializer.serializedSize(partitionUpdate, version);
                this.size = size;
            }
            return size;
        }
    }

    /**
     * Collects finalized partition updates
     */
    public static class PartitionUpdateCollector
    {
        private final ImmutableMap.Builder<TableId, PartitionUpdate> modifications = new ImmutableMap.Builder<>();
        private final String keyspaceName;
        private final DecoratedKey key;
        private final long approxCreatedAtNanos = approxTime.now();
        private boolean empty = true;

        private PotentialTxnConflicts potentialTxnConflicts;

        public PartitionUpdateCollector(String keyspaceName, DecoratedKey key)
        {
            this(keyspaceName, key, PotentialTxnConflicts.DISALLOW);
        }

        public PartitionUpdateCollector(String keyspaceName, DecoratedKey key, PotentialTxnConflicts potentialTxnConflicts)
        {
            this.keyspaceName = keyspaceName;
            this.key = key;
            this.potentialTxnConflicts = potentialTxnConflicts;
        }

        public PartitionUpdateCollector add(PartitionUpdate partitionUpdate)
        {
            assert partitionUpdate != null : "Null updates are not allowed";
            assert partitionUpdate.partitionKey().getPartitioner() == key.getPartitioner(): String.format("Update to key %s with partitioner %s (%s) had an update (%s) with a different partitioner! %s (%s)",
                                                                                                          key,
                                                                                                          key.getPartitioner(), key.getPartitioner().getClass(),
                                                                                                          partitionUpdate,
                                                                                                          partitionUpdate.partitionKey().getPartitioner(), partitionUpdate.partitionKey().getPartitioner().getClass());
            // note that ImmutableMap.Builder only allows put:ing the same key once, it will fail during build() below otherwise
            modifications.put(partitionUpdate.metadata().id, partitionUpdate);
            empty = false;
            return this;
        }

        public DecoratedKey key()
        {
            return key;
        }

        public String getKeyspaceName()
        {
            return keyspaceName;
        }

        public boolean isEmpty()
        {
            return empty;
        }

        public Mutation build()
        {
            return new Mutation(keyspaceName, key, modifications.build(), approxCreatedAtNanos, potentialTxnConflicts);
        }
    }
}