topk.cql

CREATE OR REPLACE FUNCTION topkState ( state tuple<int, map<text, int>>, val text )
  CALLED ON NULL INPUT RETURNS tuple<int, map<text, int>> LANGUAGE java AS '
java.util.Map<String, Integer> m = state.getMap(1, String.class, Integer.class);
int current = m.get(val) == null? 1 : m.get(val) + 1;
if (val != null) {
    state.setInt(0, state.getInt(0) + 1);
    if (state.getInt(0) < 100) {
        m.put(val, current);
    }
    else {
        String min = m.keySet().iterator().next();
        for(String e : m.keySet()) {
            if(m.get(e) < m.get(min)) min = e;
        }
        m.put(val, m.get(min) + 1);
        m.remove(min);
    }
}
state.setMap(1, m);
return state;';

CREATE OR REPLACE FUNCTION topkFinal (state tuple<int, map<text, int>>)
  CALLED ON NULL INPUT RETURNS map<text, int> LANGUAGE java AS 
  'return state.getMap(1, String.class, Integer.class);';

CREATE AGGREGATE IF NOT EXISTS topk ( text ) 
  SFUNC topkState STYPE tuple<int, map<text, int>> FINALFUNC topkFinal INITCOND (0,{});