src/java/org/apache/cassandra/db/Table.java
/**
     * This method adds the row to the Commit Log associated with this table.
     * Once this happens the data associated with the individual column families
     * is also written to the column family store's memtable.
    */
This method adds the row to the Commit Log associated with this table.
Once this happens the data associated with the individual column families
is also written to the column family store's memtable.
This method adds the row to the Commit Log associated with this table . Once this happens the data associated with the individual column families is also written to the column family store ' s memtable .
public void apply(RowMutation mutation, boolean writeCommitLog, boolean updateIndexes) throws IOException
    {
        if (logger.isDebugEnabled())
            logger.debug("applying mutation of row {}", ByteBufferUtil.bytesToHex(mutation.key()));
        switchLock.readLock().lock();
        try
        {
            if (writeCommitLog)
                CommitLog.instance.add(mutation);
            boolean inTransaction;
            if (mutation.getColumnFamilies().size() == 0 || mutation.getColumnFamilies().iterator().next().columns.size() == 0) {
                inTransaction = false;
            } else {
                IColumn firstColumn = mutation.getColumnFamilies().iterator().next().columns.iterator().next();
                if (firstColumn instanceof SuperColumn) {
                    if (firstColumn.getSubColumns().size() == 0) {
                        inTransaction = false;
                    } else {
                        inTransaction = firstColumn.getSubColumns().iterator().next().transactionCoordinatorKey() != null;
                    }
                } else {
                    inTransaction = firstColumn.transactionCoordinatorKey() != null;
                }
            }
            if (!inTransaction) {
                AppliedOperations.addAppliedOp(mutation.key(), mutation.extractTimestamp());
            }
            long earliestValidTime;
            if (ShortNodeId.getLocalDC() == VersionUtil.extractDatacenter(mutation.extractTimestamp())) {
                earliestValidTime = mutation.extractTimestamp();
            } else {
                earliestValidTime = LamportClock.getVersion();
            }
            for (ColumnFamily cf : mutation.getColumnFamilies()) {
                for (IColumn column : cf.columns) {
                   if (column instanceof Column) {
                       column.setEarliestValidTime(earliestValidTime);
                   } else {
                       assert column instanceof SuperColumn;
                       for (IColumn subcolumn : column.getSubColumns()) {
                           assert subcolumn instanceof Column;
                           subcolumn.setEarliestValidTime(earliestValidTime);
                       }
                   }
                }
            }
            DecoratedKey<?> key = StorageService.getPartitioner().decorateKey(mutation.key());
            for (ColumnFamily cf : mutation.getColumnFamilies())
            {
                ColumnFamilyStore cfs = columnFamilyStores.get(cf.id());
                if (cfs == null)
                {
                    logger.error("Attempting to mutate non-existant column family " + cf.id());
                    continue;
                }
                SortedSet<ByteBuffer> mutatedIndexedColumns = null;
                if (updateIndexes)
                {
                    for (ByteBuffer column : cfs.indexManager.getIndexedColumns())
                    {
                        if (cf.getColumnNames().contains(column) || cf.isMarkedForDelete())
                        {
                            if (mutatedIndexedColumns == null)
                                mutatedIndexedColumns = new TreeSet<ByteBuffer>();
                            mutatedIndexedColumns.add(column);
                            if (logger.isDebugEnabled())
                            {
                                ByteBuffer value = cf.getColumn(column) == null ? null : cf.getColumn(column).value(); 
                                logger.debug(String.format("mutating indexed column %s value %s",
                                                           cf.getComparator().getString(column),
                                                           value == null ? "null" : ByteBufferUtil.bytesToHex(value)));
                            }
                        }
                    }
                }
                if (mutatedIndexedColumns == null)
                {
                    cfs.apply(key, cf);
                    continue;
                }
                synchronized (indexLockFor(mutation.key()))
                {
                    ColumnFamily oldIndexedColumns = readCurrentIndexedColumns(key, cfs, mutatedIndexedColumns);
                    logger.debug("Pre-mutation index row is {}", oldIndexedColumns);
                    ignoreObsoleteMutations(cf, mutatedIndexedColumns, oldIndexedColumns);
                    cfs.apply(key, cf);
                    cfs.indexManager.applyIndexUpdates(mutation.key(), cf, mutatedIndexedColumns, oldIndexedColumns);
                }
            }
        }
        finally
        {
            switchLock.readLock().unlock();
        }
    }
public void apply ( RowMutation mutation , boolean writeCommitLog , boolean updateIndexes ) throws IOException { if ( logger . isDebugEnabled ( ) ) logger . debug ( "applying mutation of row {}" , ByteBufferUtil . bytesToHex ( mutation . key ( ) ) ) ; switchLock . readLock ( ) . lock ( ) ; try { if ( writeCommitLog ) CommitLog . instance . add ( mutation ) ; boolean inTransaction ; if ( mutation . getColumnFamilies ( ) . size ( ) == 0 || mutation . getColumnFamilies ( ) . iterator ( ) . next ( ) . columns . size ( ) == 0 ) { inTransaction = false ; } else { IColumn firstColumn = mutation . getColumnFamilies ( ) . iterator ( ) . next ( ) . columns . iterator ( ) . next ( ) ; if ( firstColumn instanceof SuperColumn ) { if ( firstColumn . getSubColumns ( ) . size ( ) == 0 ) { inTransaction = false ; } else { inTransaction = firstColumn . getSubColumns ( ) . iterator ( ) . next ( ) . transactionCoordinatorKey ( ) != null ; } } else { inTransaction = firstColumn . transactionCoordinatorKey ( ) != null ; } } if ( ! inTransaction ) { AppliedOperations . addAppliedOp ( mutation . key ( ) , mutation . extractTimestamp ( ) ) ; } long earliestValidTime ; if ( ShortNodeId . getLocalDC ( ) == VersionUtil . extractDatacenter ( mutation . extractTimestamp ( ) ) ) { earliestValidTime = mutation . extractTimestamp ( ) ; } else { earliestValidTime = LamportClock . getVersion ( ) ; } for ( ColumnFamily cf : mutation . getColumnFamilies ( ) ) { for ( IColumn column : cf . columns ) { if ( column instanceof Column ) { column . setEarliestValidTime ( earliestValidTime ) ; } else { assert column instanceof SuperColumn ; for ( IColumn subcolumn : column . getSubColumns ( ) ) { assert subcolumn instanceof Column ; subcolumn . setEarliestValidTime ( earliestValidTime ) ; } } } } DecoratedKey < ? > key = StorageService . getPartitioner ( ) . decorateKey ( mutation . key ( ) ) ; for ( ColumnFamily cf : mutation . getColumnFamilies ( ) ) { ColumnFamilyStore cfs = columnFamilyStores . get ( cf . id ( ) ) ; if ( cfs == null ) { logger . error ( "Attempting to mutate non-existant column family " + cf . id ( ) ) ; continue ; } SortedSet < ByteBuffer > mutatedIndexedColumns = null ; if ( updateIndexes ) { for ( ByteBuffer column : cfs . indexManager . getIndexedColumns ( ) ) { if ( cf . getColumnNames ( ) . contains ( column ) || cf . isMarkedForDelete ( ) ) { if ( mutatedIndexedColumns == null ) mutatedIndexedColumns = new TreeSet < ByteBuffer > ( ) ; mutatedIndexedColumns . add ( column ) ; if ( logger . isDebugEnabled ( ) ) { ByteBuffer value = cf . getColumn ( column ) == null ? null : cf . getColumn ( column ) . value ( ) ; logger . debug ( String . format ( "mutating indexed column %s value %s" , cf . getComparator ( ) . getString ( column ) , value == null ? "null" : ByteBufferUtil . bytesToHex ( value ) ) ) ; } } } } if ( mutatedIndexedColumns == null ) { cfs . apply ( key , cf ) ; continue ; } synchronized ( indexLockFor ( mutation . key ( ) ) ) { ColumnFamily oldIndexedColumns = readCurrentIndexedColumns ( key , cfs , mutatedIndexedColumns ) ; logger . debug ( "Pre-mutation index row is {}" , oldIndexedColumns ) ; ignoreObsoleteMutations ( cf , mutatedIndexedColumns , oldIndexedColumns ) ; cfs . apply ( key , cf ) ; cfs . indexManager . applyIndexUpdates ( mutation . key ( ) , cf , mutatedIndexedColumns , oldIndexedColumns ) ; } } } finally { switchLock . readLock ( ) . unlock ( ) ; } }
This method adds the row to the Commit Log associated with this table.
This method adds the row to the Commit Log associated with this table .
// write the mutation to the commitlog and memtables //We consider the operation applied now //WL TODO Perhaps we should move this into ColumnFamilyStore.apply and do it individually for all CFs? //Determine if this is part of a transaction, if it isn't, update AppliedOps, if it is, that's done in the Coordinator/Cohort //Set the earliest_valid_time for all columns in this update //if the update originates from this node, it's earliest valid time == timestamp //otherwise it's the current logical time // can't actually use validator to print value here, because we overload value // for deletion timestamp as well (which may not be a well-formed value for the column type) // may be null on row-level deletion // Sharding the lock is insufficient to avoid contention when there is a "hot" row, e.g., for // hint writes when a node is down (keyed by target IP).  So it is worth special-casing the // no-index case to avoid the synchronization. // else mutatedIndexedColumns != null // with the raw data CF, we can just apply every update in any order and let // read-time resolution throw out obsolete versions, thus avoiding read-before-write. // but for indexed data we need to make sure that we're not creating index entries // for obsolete writes. // ignore full index memtables -- we flush those when the "master" one is full