class EntryParser implements Function<Entry<Key,Value>,Tuple2<String,IndexInfo>> {

    @Override
    public Tuple2<String,IndexInfo> apply(Entry<Key,Value> entry) {
        IndexInfo info = new IndexInfo();
        try {
            info.readFields(new DataInputStream(new ByteArrayInputStream(entry.getValue().get())));
        } catch (IOException e) {
            return null;
        }
        String date = entry.getKey().getColumnQualifier().toString();
        
        if (log.isTraceEnabled()) {
            log.trace("Adding " + currNode + " to " + entry.getKey() + " ");
            for (IndexMatch match : info.uids()) {
                log.trace(date + " " + match.getUid().split("\u0000")[1]);
            }
        }
        
        if (!skipNodeDelay && Union.isDay(date) && info.uids().isEmpty()) {
            
            if (isDelayedPredicate(currNode)) {
                if (log.isTraceEnabled()) {
                    log.trace("not delaying " + currNode + " because it is already delayed" + currNode.jjtGetParent() + "<- parent "
                                    + JexlStringBuildingVisitor.buildQuery(currNode) + " " + date + " " + info.uids.size());
                }
                info.applyNode(currNode);
            } else if (null != indexOnlyFields && indexOnlyFields.contains(fieldName)) {
                if (log.isTraceEnabled()) {
                    log.trace("not delaying " + currNode + " because it is index only");
                }
                info.applyNode(currNode);
            } else {
                if (log.isTraceEnabled()) {
                    log.trace("delaying " + currNode + " because it is already delayed" + currNode.jjtGetParent() + "<- parent "
                                    + JexlStringBuildingVisitor.buildQuery(currNode) + " " + date + " " + info.uids.size());
                }
                info.applyNode(ASTDelayedPredicate.create(JexlNodeFactory.buildEQNode(fieldName, literal)));
            }
        } else {
            if (log.isTraceEnabled()) {
                log.trace(date + " Size is " + info.uids.size() + " count is " + info.count);
            }
            info.applyNode(currNode);
        }
        return Tuples.tuple(entry.getKey().getColumnQualifier().toString(), info);
    }

    public  EntryParser(ASTEQNode node, String fieldName, String literal);
    public  EntryParser(String fieldName, String literal, boolean skipNodeDelay);
    public  EntryParser(ASTEQNode node, String fieldName, String literal, Set<String> indexOnlyFields);

}