Delete-Insert Trigger

DeleteInsert.java

/*
javac -cp "/home/digger/dse/resources/cassandra/lib/*" -d target/ src/hessian/invest/DeleteInsert.java
jar cf DeleteInsert.jar -C target/ hessian
cp DeleteInsert.jar ~/dse/resources/cassandra/conf/triggers/
nodetool reloadtriggers

CREATE TRIGGER deleteInsert ON ks.tbl USING 'hessian.invest.DeleteInsert';
*/

package hessian.invest;

import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.Collections;
import org.apache.cassandra.triggers.ITrigger;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.cql3.*;


public class DeleteInsert implements ITrigger {
    public Collection<Mutation> augmentNonBlocking(Partition partition)
    {
	PartitionUpdate.SimpleBuilder pusb = PartitionUpdate.simpleBuilder(partition.metadata(),
									   partition.partitionKey().getKey());

	// Get TableMetadata, ColumnMetadata, etc
	TableMetadata tmeta = partition.metadata();
	long mintimestamp = Long.MAX_VALUE;

	// Iterate through Rows in Partition
	UnfilteredRowIterator iter = partition.unfilteredIterator();
	while (iter.hasNext()){
	    Unfiltered u = iter.next();
	    // Only look at Rows (not Range Tombstone Marker)
	    if (!u.isRow()) continue;
	    Row row = (Row)u;
	    long tts = row.primaryKeyLivenessInfo().timestamp();
	    if (tts < mintimestamp)
		mintimestamp = tts;
	}
	if (mintimestamp < Long.MAX_VALUE) {
	    pusb.timestamp(mintimestamp - 1);
	    pusb.delete();
	}

	return Collections.singletonList(pusb.buildAsMutation());
    }
}