natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

TTL support for BulkOutputFormat. 

Patch by Samarth Gahire and brandonwilliams, reviewed by brandonwilliams
for CASSANDRA-3754
This commit is contained in:
Brandon Williams 2012-02-06 12:20:52 -06:00
parent 6c2ded6e98
commit fdd3721fba
1 changed files with 12 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/java/org/apache/cassandra/hadoop/BulkRecordWriter.java
View File

@ -148,7 +148,12 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
if (colType == ColType.COUNTER)
writer.addCounterColumn(column.name, column.value.getLong());
else
writer.addColumn(column.name, column.value, column.timestamp);
{
if(0 == column.ttl)
writer.addColumn(column.name, column.value, column.timestamp);
else
writer.addExpiringColumn(column.name, column.value, column.timestamp, column.ttl, System.currentTimeMillis() + (column.ttl * 1000));
}
}
}
else
@ -156,11 +161,15 @@ implements org.apache.hadoop.mapred.RecordWriter<ByteBuffer,List<Mutation>>
if (colType == ColType.COUNTER)
writer.addCounterColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value.getLong());
else
writer.addColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp);
{
if(0 == mut.getColumn_or_supercolumn().column.ttl)
writer.addColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp);
else
writer.addExpiringColumn(mut.getColumn_or_supercolumn().column.name, mut.getColumn_or_supercolumn().column.value, mut.getColumn_or_supercolumn().column.timestamp, mut.getColumn_or_supercolumn().column.ttl, System.currentTimeMillis() + (mut.getColumn_or_supercolumn().column.ttl * 1000));
}
}
}
}
@Override
public void close(TaskAttemptContext context) throws IOException, InterruptedException
{