package org.apache.cassandra.logger.build;

import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.composites.CellName;
import org.apache.cassandra.db.composites.CellNames;
import org.apache.cassandra.db.marshal.TimestampType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.logger.log.LogEntry;
import org.apache.commons.lang3.StringUtils;

import java.util.Collection;
import java.util.Date;

public class LogMutationBuilder {

    private String keyspace;
    private String columnFamily;

    public LogMutationBuilder(String keyspace, String columnFamily) {
        this.keyspace = keyspace;
        this.columnFamily = columnFamily;
    }

    public Mutation build(LogEntry logEntry) {
        long timestamp = new Date().getTime();
        
        Mutation mutation = new Mutation(keyspace, UUIDType.instance.decompose(logEntry.getId()));
        
        mutation.add(columnFamily, toCellName("time"), TimestampType.instance.decompose(logEntry.getTime()), timestamp);        
        mutation.add(columnFamily, toCellName("logged_keyspace"), UTF8Type.instance.decompose(logEntry.getLoggedKeyspace()), timestamp);
        mutation.add(columnFamily, toCellName("logged_table"), UTF8Type.instance.decompose(logEntry.getLoggedTable()), timestamp);
        mutation.add(columnFamily, toCellName("logged_key"), UTF8Type.instance.decompose(logEntry.getLoggedKey()), timestamp);
        mutation.add(columnFamily, toCellName("updated_columns"), UTF8Type.instance.decompose(toCommaSeparatedString(logEntry.getUpdatedColumns())), timestamp);
        mutation.add(columnFamily, toCellName("operation"), UTF8Type.instance.decompose(logEntry.getOperation().getValue()), timestamp);
        
        return mutation;
    }

    private CellName toCellName(String name) {
        return CellNames.simpleSparse(new ColumnIdentifier(name, false));
    }
    
    private String toCommaSeparatedString(Collection<String> list) {
        return StringUtils.join(list, ',');
    }
}
@@ -327,20 +327,21 @@
 marshal.
-UTF8
+Ascii
 Type;%0Aim
@@ -988,25 +988,17 @@
 Id()));%0A
-        
 %0A
+
         
@@ -1180,28 +1180,29 @@
 keyspace%22), 
-UTF8
+Ascii
 Type.instanc
@@ -1315,28 +1315,29 @@
 ed_table%22), 
-UTF8
+Ascii
 Type.instanc
@@ -1445,28 +1445,29 @@
 gged_key%22), 
-UTF8
+Ascii
 Type.instanc
@@ -1578,28 +1578,29 @@
 _columns%22), 
-UTF8
+Ascii
 Type.instanc
@@ -1742,12 +1742,13 @@
 %22), 
-UTF8
+Ascii
 Type

Use AsciiType instead of UTF8Type since we're dealing with names that are not supposed to have localization.