### Question:
SumTopTbl implements Tbl { public List<Map.Entry<CharSequence, GenericData.Record>> filter(Map<CharSequence, GenericData.Record> curValue) { List<Map.Entry<CharSequence, GenericData.Record>> elements = Lists.newArrayList(); for (Map.Entry<CharSequence, GenericData.Record> e : curValue.entrySet()) { if (keep == null) { elements.add(e); } else { GenericData.Record v = e.getValue(); Scriptable evalScope = ctx.newObject(scope); evalScope.setPrototype(scope); evalScope.setParentScope(null); for (int i = 0; i < v.getSchema().getFields().size(); i++) { evalScope.put(v.getSchema().getFields().get(i).name(), evalScope, v.get(i)); } Boolean kept = (Boolean) keep.exec(ctx, evalScope); if (kept.booleanValue()) { elements.add(e); } } } return elements; } SumTopTbl(Map<String, String> values, Map<String, Object> options); @Override int arity(); @Override SchemaProvider getSchemas(ObsDescriptor od, int outputId, int aggIdx); @Override void initialize(SchemaProvider provider); @Override void add(Obs obs); @Override GenericData.Record getValue(); @Override GenericData.Record merge(
      GenericData.Record current,
      GenericData.Record next); List<Map.Entry<CharSequence, GenericData.Record>> filter(Map<CharSequence, GenericData.Record> curValue); List<Map.Entry<CharSequence, GenericData.Record>> sort(
      List<Map.Entry<CharSequence, GenericData.Record>> elements); @Override List<GenericData.Record> finalize(GenericData.Record input);  }### Answer:
@Test public void testFilter() { Map<String, Object> opts = Maps.newHashMap(); opts.put("by", "key"); opts.put("order", "a * b"); opts.put("limit", 1); opts.put("keep", "a > 0 && b > 0"); SumTopTbl stt = new SumTopTbl(values, opts); ObsDescriptor od = new AvroObsDescriptor(schema); SchemaProvider sp = stt.getSchemas(od, 0, 0); stt.initialize(sp); List<Map.Entry<CharSequence, GenericData.Record>> elem = stt.filter(cv); assertEquals(1, elem.size()); assertEquals("x", elem.get(0).getValue().get("key")); }