LWW and Time Travel

lww_timetravel.cql

CREATE KEYSPACE lab WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'};
use lab;

CREATE OR REPLACE FUNCTION lww_accumulate(state MAP<INT,INT>, mutid INT, val INT)
CALLED ON NULL INPUT
RETURNS MAP<INT,INT>
LANGUAGE java
AS '
state.put(mutid, val);
return state;
';

CREATE OR REPLACE FUNCTION lww_final(state MAP<INT,INT>)
CALLED ON NULL INPUT
RETURNS INT
LANGUAGE java
AS '
List<Integer> keys = new ArrayList<Integer>(state.keySet());
if (keys.size() < 1) { return -1; }
for (int i = 0; i < keys.size(); i++) {
    int k = keys.get(i);
    if (k < 0) {
        state.remove(k);
        state.remove(k * -1);
    }
}
if (state.keySet().size() < 1)
    return -1;
keys = new ArrayList<Integer>(state.keySet());
Collections.sort(keys);
return state.get(keys.get(keys.size()-1));
';

CREATE OR REPLACE AGGREGATE lww(INT, INT) 
SFUNC lww_accumulate
STYPE MAP<INT,INT>
FINALFUNC lww_final
INITCOND {};




CREATE OR REPLACE FUNCTION asof_accumulate(state MAP<INT,INT>, mutid INT, val INT, max INT)
CALLED ON NULL INPUT
RETURNS MAP<INT,INT>
LANGUAGE java
AS '
if ((mutid != null) && (mutid <= max)) {
    if (state.isEmpty()) {
        state.put(mutid, val);
    }
    else {
        Integer k = state.keySet().iterator().next();
        Integer v = state.get(k);
        if (mutid > k) {
            state.clear();
            if (val != null) {
                state.put(mutid, val);
            }
        }
    }
}
return state;
';

CREATE OR REPLACE FUNCTION asof_final(state MAP<INT,INT>)
CALLED ON NULL INPUT
RETURNS INT
LANGUAGE java
AS '
if (state.isEmpty()) {
    return null;
}
return state.values().iterator().next();
';

CREATE OR REPLACE AGGREGATE asof(INT,INT,INT)
SFUNC asof_accumulate
STYPE MAP<INT,INT>
FINALFUNC asof_final
INITCOND {};



CREATE TABLE log (
    pkey int,
    ccol int,
    mutid int,
    x int,
    y int,
    PRIMARY KEY (pkey, ccol, mutid)
);

INSERT INTO log(pkey,ccol,mutid,x,y) VALUES (1,1,10,1,100);
INSERT INTO log(pkey,ccol,mutid,x,y) VALUES (1,1,20,2,200);
INSERT INTO log(pkey,ccol,mutid,x,y) VALUES (1,1,30,3,300);

SELECT pkey, ccol, Lww(mutid,x) AS x, Lww(mutid,y) AS y FROM log GROUP BY pkey, ccol;
SELECT pkey, ccol, Asof(mutid,x,11) AS x, Asof(mutid,y,11) AS y FROM log GROUP BY pkey, ccol;

INSERT INTO log(pkey,ccol,mutid) VALUES (1,1,-30); // deletes mutation 30
SELECT pkey, ccol, Lww(mutid,x) AS x, Lww(mutid,y) AS y FROM log GROUP BY pkey, ccol;