import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
public void addPoint(String key, List<SeriesValue> values) {
    boolean nullKey = false;
    try {
        registerKey(key);
        MutationBatch m = keyspace.prepareMutationBatch();
        ColumnListMutation<String> mut = m.withRow(columnFamily, key);
        for (SeriesValue value : values) {
            if (value.getColumn() == null) nullKey = true;
            else mut.putColumn(value.getColumn(), SeriesValueCodec.encodeValue(value), null);
        }
        m.execute();
    } catch (ConnectionException ce) {
        throw RaptureExceptionFactory.create(HttpURLConnection.HTTP_INTERNAL_ERROR, messageCatalog.getMessage("DbCommsError"), ce);
    } catch (UnsupportedEncodingException e) {
        throw RaptureExceptionFactory.create(HttpURLConnection.HTTP_INTERNAL_ERROR, messageCatalog.getMessage("BadSeriesValue"), e);
    }
    if (nullKey) throw RaptureExceptionFactory.create(HttpURLConnection.HTTP_BAD_REQUEST, messageCatalog.getMessage("BadKey"));
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void deleteOldColumns(AstyanaxTable table, String blobId, ColumnList<Composite> columns, long timestamp) {
    for (AstyanaxStorage storage : table.getWriteStorage()) {
        BlobPlacement placement = (BlobPlacement) storage.getPlacement();

        // Any columns with a timestamp older than the one we expect must be from an old version
        // of the blob.  This should be rare, but if it happens clean up and delete the old data.
        MutationBatch mutation = placement.getKeyspace().prepareMutationBatch(ConsistencyLevel.CL_ANY);
        ColumnListMutation<Composite> row = mutation.withRow(
                placement.getBlobColumnFamily(), storage.getRowKey(blobId));
        boolean found = false;
        for (Column<Composite> column : columns) {
            if (column.getTimestamp() < timestamp) {
                if (ColumnGroup.B.name().equals(column.getName().get(0, AsciiSerializer.get()))) {
                    int chunkId = column.getName().get(1, IntegerSerializer.get());
                    row.deleteColumn(getColumn(ColumnGroup.B, chunkId))
                            .deleteColumn(getColumn(ColumnGroup.Z, chunkId));
                    found = true;
                }
            }
        }
        if (found) {
            execute(mutation);
        }
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
public void createInstanceEntry(RaigadInstance instance) throws Exception {
    logger.info("Creating new instance entry");

    String key = getRowKey(instance);
    // If the key exists throw exception
    if (getInstance(instance.getApp(), instance.getDC(), instance.getId()) != null) {
        logger.info(String.format("Key already exists: %s", key));
        return;
    }

    // Grab the lock
    getLock(instance);

    MutationBatch mutationBatch = bootKeyspace.prepareMutationBatch();
    ColumnListMutation<String> columnListMutation = mutationBatch.withRow(CF_INSTANCES, key);
    columnListMutation.putColumn(CN_CLUSTER, instance.getApp(), null);
    columnListMutation.putColumn(CN_AZ, instance.getAvailabilityZone(), null);
    columnListMutation.putColumn(CN_INSTANCEID, instance.getInstanceId(), null);
    columnListMutation.putColumn(CN_HOSTNAME, instance.getHostName(), null);
    columnListMutation.putColumn(CN_IP, instance.getHostIP(), null);
    columnListMutation.putColumn(CN_LOCATION, instance.getDC(), null);
    columnListMutation.putColumn(CN_ASGNAME, instance.getAsg(), null);
    columnListMutation.putColumn(CN_UPDATETIME, TimeUUIDUtils.getUniqueTimeUUIDinMicros(), null);

    mutationBatch.execute();
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Override
    public void tagId(String id, Map<String, String> tags) throws IndexerException  {
        MutationBatch mb = keyspace.prepareMutationBatch();
        
        ColumnListMutation<String> idRow = mb.withRow(dataCf, id);
        UUID uuid = TimeUUIDUtils.getUniqueTimeUUIDinMicros();
        for (Map.Entry<String, String> tag : tags.entrySet()) {
            String rowkey = tag.getKey() + "=" + tag.getValue();
            System.out.println("Rowkey: " + rowkey);
            mb.withRow(indexCf, tag.getKey() + "=" + tag.getValue())
              .putEmptyColumn(new IndexEntry(id, uuid));
//            idRow.putColumn(tag.getKey(), tag.getValue());
        }
        
        try {
            mb.execute();
        } catch (ConnectionException e) {
            throw new IndexerException("Failed to store tags : " + tags + " for id " + id, e);
        }
    }
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
public void updateRow(String key, RowData rowData) throws PaasException {
    LOG.info("Update row: " + rowData.toString());
    invariant();
    
    MutationBatch mb = keyspace.prepareMutationBatch();
    if (rowData.hasSchemalessRows()) {
        ColumnListMutation<ByteBuffer> mbRow = mb.withRow(this.columnFamily, serializers.keyAsByteBuffer(key));
        for (Entry<String, Map<String, String>> row : rowData.getSrows().getRows().entrySet()) {
            for (Entry<String, String> column : row.getValue().entrySet()) {
                mbRow.putColumn(serializers.columnAsByteBuffer(column.getKey()),  
                                serializers.valueAsByteBuffer(column.getKey(), column.getValue()));
            }
        }
    }
    
    try {
        mb.execute();
    } catch (ConnectionException e) {
        throw new PaasException(
                String.format("Failed to update row '%s' in column family '%s.%s'" , 
                              key, this.keyspace.getKeyspaceName(), this.columnFamily.getName()),
                e);
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Override
public void updateColumn(String key, String column, String value) throws NotFoundException, PaasException {
    LOG.info("Update row");
    invariant();
    
    MutationBatch mb = keyspace.prepareMutationBatch();
    ColumnListMutation<ByteBuffer> mbRow = mb.withRow(this.columnFamily, serializers.keyAsByteBuffer(key));
    mbRow.putColumn(serializers.columnAsByteBuffer(column),  
                    serializers.valueAsByteBuffer(column, value));
    
    try {
        mb.execute();
    } catch (ConnectionException e) {
        throw new PaasException(
                String.format("Failed to update row '%s' in column family '%s.%s'" , 
                              key, this.keyspace.getKeyspaceName(), this.columnFamily.getName()),
                e);
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Override
public void deleteColumn(String key, String column) throws PaasException {
    LOG.info("Update row");
    invariant();
    
    MutationBatch mb = keyspace.prepareMutationBatch();
    ColumnListMutation<ByteBuffer> mbRow = mb.withRow(this.columnFamily, serializers.keyAsByteBuffer(key));
    mbRow.deleteColumn(serializers.columnAsByteBuffer(column));
    try {
        mb.execute();
    } catch (ConnectionException e) {
        throw new PaasException(
                String.format("Failed to update row '%s' in column family '%s.%s'" , 
                              key, this.keyspace.getKeyspaceName(), this.columnFamily.getName()),
                e);
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
public void writeMetadata(Table<Locator, String, String> metaTable) throws ConnectionException {
    ColumnFamily cf = CassandraModel.CF_METRICS_METADATA;
    Timer.Context ctx = Instrumentation.getBatchWriteTimerContext(CassandraModel.CF_METRICS_METADATA_NAME);
    MutationBatch batch = keyspace.prepareMutationBatch();

    try {
        for (Locator locator : metaTable.rowKeySet()) {
            Map<String, String> metaRow = metaTable.row(locator);
            ColumnListMutation<String> mutation = batch.withRow(cf, locator);

            for (Map.Entry<String, String> meta : metaRow.entrySet()) {
                mutation.putColumn(meta.getKey(), meta.getValue(), StringMetadataSerializer.get(), null);
            }
        }
        try {
            batch.execute();
        } catch (ConnectionException e) {
            Instrumentation.markWriteError(e);
            log.error("Connection exception persisting metadata", e);
            throw e;
        }
    } finally {
        ctx.stop();
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Override
public MutationBatch write( final ApplicationScope applicationScope, final MvccEntity entity ) {
    Preconditions.checkNotNull( applicationScope, "applicationScope is required" );
    Preconditions.checkNotNull( entity, "entity is required" );

    final UUID colName = entity.getVersion();
    final Id entityId = entity.getId();

    return doWrite( applicationScope, entityId, new RowOp() {
        @Override
        public void doOp( final ColumnListMutation<UUID> colMutation ) {
                colMutation.putColumn( colName, getEntitySerializer()
                        .toByteBuffer( new EntityWrapper( entity.getStatus(), entity.getEntity() ) ) );
        }
    } );
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Override
protected void doWrite(Publisher source, List<ChannelSchedule> blocks) throws WriteException {
    try {
        MutationBatch batch = keyspace.prepareMutationBatch().withConsistencyLevel(writeCl);
        for (ChannelSchedule block : blocks) {
            ColumnListMutation<String> rowMutation = batch.withRow(cf, key(source, block));
            for (ItemAndBroadcast entry : block.getEntries()) {
                rowMutation.putColumn(
                        entry.getBroadcast().getSourceId(),
                        serializer.serialize(entry)
                );
            }
            rowMutation.putColumn(IDS_COL, Joiner.on(',').join(broadcastIds(block)));
            rowMutation.putColumn(UPDATED_COL, clock.now().toDate());
        }
        try {
            batch.execute();
        } catch (ConnectionException ce) {
            throw new WriteException(ce);
        }
    } catch (WriteException | RuntimeException e) {
        throw e;
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void handleChildRefColumn(
        ColumnListMutation<String> mutation,
        Id id,
        ContentProtos.Content msg,
        Descriptors.FieldDescriptor fd
) {
    if (msg.getRepeatedFieldCount(fd) == 1) {
        CommonProtos.Reference cr = (CommonProtos.Reference) msg.getRepeatedField(fd, 0);
        ContentProtos.Content col = ContentProtos.Content.newBuilder()
                .addRepeatedField(fd, cr)
                .build();
        addColumnToBatch(
                mutation,
                id,
                String.valueOf(cr.getId()),
                col.toByteArray()
        );
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void handleUpcomingContentColumn(
        ColumnListMutation<String> mutation,
        Id id,
        ContentProtos.Content msg,
        Descriptors.FieldDescriptor fd
) {
    if (msg.getRepeatedFieldCount(fd) == 1) {
        ContentProtos.ItemAndBroadcastRef uc =
                (ContentProtos.ItemAndBroadcastRef) msg.getRepeatedField(fd, 0);
        ContentProtos.Content col = ContentProtos.Content.newBuilder()
                .addRepeatedField(fd, uc)
                .build();
        addColumnToBatch(
                mutation,
                id,
                buildUpcomingContentKey(uc.getItem().getId()),
                col.toByteArray()
        );
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void handleAvailableContentColumn(
        ColumnListMutation<String> mutation,
        Id id,
        ContentProtos.Content msg,
        Descriptors.FieldDescriptor fd
) {
    if (msg.getRepeatedFieldCount(fd) == 1) {
        ContentProtos.ItemAndLocationSummary uc =
                (ContentProtos.ItemAndLocationSummary) msg.getRepeatedField(fd, 0);
        ContentProtos.Content col = ContentProtos.Content.newBuilder()
                .addRepeatedField(fd, uc)
                .build();
        addColumnToBatch(
                mutation,
                id,
                buildAvailableContentKey(uc.getItem().getId()),
                col.toByteArray()
        );
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void handleItemSummariesColumn(
        ColumnListMutation<String> mutation,
        Id id,
        ContentProtos.Content msg,
        Descriptors.FieldDescriptor fd
) {
    if (msg.getRepeatedFieldCount(fd) == 1) {
        ContentProtos.ItemSummary uc = (ContentProtos.ItemSummary) msg.getRepeatedField(fd, 0);
        ContentProtos.Content col = ContentProtos.Content.newBuilder()
                .addRepeatedField(fd, uc)
                .build();
        addColumnToBatch(
                mutation,
                id,
                buildItemSummaryKey(uc.getItemRef().getId()),
                col.toByteArray()
        );
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void handleBroadcastColumn(
        ColumnListMutation<String> mutation,
        Id id,
        ContentProtos.Content msg,
        Iterable<String> staleBroadcastSourceIds,
        Descriptors.FieldDescriptor fd
) {
    int broadcastCount = msg.getRepeatedFieldCount(fd);
    for (int i = 0; i < broadcastCount; i++) {
        ContentProtos.Broadcast uc = (ContentProtos.Broadcast) msg.getRepeatedField(fd, i);
        ContentProtos.Content col = ContentProtos.Content.newBuilder()
                .addRepeatedField(fd, uc)
                .build();
        addColumnToBatch(
                mutation,
                id,
                buildBroadcastKey(uc.getSourceId()),
                col.toByteArray()
        );
    }

    for (String staleBroadcastSourceId : staleBroadcastSourceIds) {
        removeColumn(mutation, buildBroadcastKey(staleBroadcastSourceId));
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private ColumnListMutation<C> row() {
    // If this is the first column in the row or after a flush, setup the mutation and row objects.
    if (_mutation != mutation()) {
        flushIf(_numRows >= _maxRows);
        _mutation = mutation();
        _row = _mutation.withRow(_columnFamily, _rowKey);
        _numRows++;
        if (_newRowCallback != null) {
            _newRowCallback.apply(this);
        }
    }
    return _row;
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Timed (name = "bv.emodb.sorAstyanaxDataWriterDAO.storeCompactedDeltas", absolute = true)
@Override
public void storeCompactedDeltas(Table tbl, String key, List<History> audits, WriteConsistency consistency) {
    checkNotNull(tbl, "table");
    checkNotNull(key, "key");
    checkNotNull(audits, "audits");
    checkNotNull(consistency, "consistency");

    AstyanaxTable table = (AstyanaxTable) tbl;
    for (AstyanaxStorage storage : table.getWriteStorage()) {
        DeltaPlacement placement = (DeltaPlacement) storage.getPlacement();
        CassandraKeyspace keyspace = placement.getKeyspace();

        ByteBuffer rowKey = storage.getRowKey(key);

        MutationBatch mutation = keyspace.prepareMutationBatch(SorConsistencies.toAstyanax(consistency));
        ColumnListMutation<UUID> rowMutation = mutation.withRow(placement.getDeltaHistoryColumnFamily(), rowKey);

        for (History history : audits) {
            rowMutation.putColumn(history.getChangeId(),
                    _changeEncoder.encodeHistory(history),
                    Ttls.toSeconds(_auditStore.getHistoryTtl(), 1, null));
        }
        execute(mutation, "store %d compacted deltas for placement %s, table %s, key %s",
                audits.size(), placement.getName(), table.getName(), key);
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
/**
 * Copies columns exactly, preserving Cassandra timestamp and ttl values.
 */
private <C> void putAll(ColumnListMutation<C> mutation, Iterable<Column<C>> columns) {
    for (Column<C> column : columns) {
        mutation.setTimestamp(column.getTimestamp())
                .putColumn(column.getName(), column.getByteBufferValue(), column.getTtl());
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
@Override
public void commit(List<History> historyList, Object rowKey) {
    if (historyList != null && !historyList.isEmpty()) {
        ColumnListMutation<UUID> historyMutation = _mutation.withRow(_columnFamily, (ByteBuffer)rowKey);
        for (History history : historyList) {
            historyMutation.putColumn(history.getChangeId(),
                    _changeEncoder.encodeHistory(history),
                    Ttls.toSeconds(_auditStore.getHistoryTtl(), 1, null));
        }
    }
}
 

import com.netflix.astyanax.ColumnListMutation; //导入依赖的package包/类
private void removeColumns(Remove action, ColumnFamily cf, MutationBatch m) {
	ColumnListMutation row = m.withRow(cf, action.getRowKey());
	
	for(byte[] name : action.getColumns()) {
		row.deleteColumn(name);
	}
}