return persistent;
}
@Override
public void put(K key, T obj) throws IOException {
Mutate mutate = new Mutate();
Schema schema = obj.getSchema();
StateManager stateManager = obj.getStateManager();
List<Field> fields = schema.getFields();
String qual;
byte[] value;
for (int i = 0; i < fields.size(); i++) {
if (!stateManager.isDirty(obj, i)) {
continue;
}
Field field = fields.get(i);
Type type = field.schema().getType();
Object o = obj.get(i);
CassandraColumn col = columnMap.get(field.name());
switch(type) {
case MAP:
if(o instanceof StatefulMap) {
@SuppressWarnings("unchecked")
StatefulMap<Utf8, ?> map = (StatefulMap<Utf8, ?>) o;
for (Entry<Utf8, State> e : map.states().entrySet()) {
Utf8 mapKey = e.getKey();
switch (e.getValue()) {
case DIRTY:
qual = mapKey.toString();
value = ByteUtils.toBytes(map.get(mapKey), field.schema().getValueType(), datumWriter);
if (col.isSuperColumn()) {
mutate.put(col.family, col.superColumn, qual, value);
} else {
mutate.put(col.family, qual, value);
}
break;
case DELETED:
qual = mapKey.toString();
if (col.isSuperColumn()) {
mutate.delete(col.family, col.superColumn, qual);
} else {
mutate.delete(col.family, qual);
}
break;
}
}
} else {
@SuppressWarnings({ "rawtypes", "unchecked" })
Set<Map.Entry> set = ((Map)o).entrySet();
for(@SuppressWarnings("rawtypes") Entry entry: set) {
qual = entry.getKey().toString();
value = ByteUtils.toBytes(entry.getValue().toString());
if (col.isSuperColumn()) {
mutate.put(col.family, col.superColumn, qual, value);
} else {
mutate.put(col.family, qual, value);
}
}
}
break;
case ARRAY:
if(o instanceof GenericArray) {
@SuppressWarnings("rawtypes")
GenericArray arr = (GenericArray) o;
int j=0;
for(Object item : arr) {
value = ByteUtils.toBytes(item.toString());
if (col.isSuperColumn()) {
mutate.put(col.family, col.superColumn, Integer.toString(j), value);
} else {
mutate.put(col.family, Integer.toString(j), value);
}
j++;
}
}
break;
default:
value = ByteUtils.toBytes(o, field.schema(), datumWriter);
if (col.isSuperColumn()) {
mutate.put(col.family, col.superColumn, col.column, value);
} else {
mutate.put(col.family, col.column, value);
}
break;
}
}
if(!mutate.isEmpty())
client.mutate(key.toString(), mutate);
}