// Path: src/main/java/org/apache/accumulo/storagehandler/predicate/compare/CompareOp.java
// public interface CompareOp {
//     /**
//      *
//      * @param comp
//      */
//     public void setPrimitiveCompare(PrimitiveCompare comp);
// 
//     /**
//      *
//      *
//      * @return PrimitiveCompare
//      */
//     public PrimitiveCompare getPrimitiveCompare();
// 
//     /**
//      *
//      *
//      * @param val
//      * @return boolean
//      */
//     public boolean accept(byte [] val);
// }
// 
// Path: src/main/java/org/apache/accumulo/storagehandler/predicate/compare/PrimitiveCompare.java
// public interface PrimitiveCompare {
// 
//     public boolean isEqual(byte[] value);
//     public boolean isNotEqual(byte[] value);
//     public boolean greaterThanOrEqual(byte[] value);
//     public boolean greaterThan(byte[] value);
//     public boolean lessThanOrEqual(byte[] value);
//     public boolean lessThan(byte[] value);
//     public boolean like(byte[] value);
//     public Object serialize(byte[] value);
//     public void init(byte[] constant);
// }


import com.google.common.collect.Lists;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.user.WholeRowIterator;
import org.apache.accumulo.storagehandler.predicate.compare.CompareOp;
import org.apache.accumulo.storagehandler.predicate.compare.PrimitiveCompare;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;
package org.apache.accumulo.storagehandler.predicate;



/**
 * Operates over a single qualifier.
 *
 * Delegates to PrimitiveCompare and CompareOpt instances for
 * value acceptance.
 *
 * The PrimitiveCompare strategy assumes a consistent value type for the same column family and qualifier.
 *
 */
public class PrimitiveComparisonFilter extends WholeRowIterator {

    public static final String FILTER_PREFIX = "accumulo.filter.compare.iterator.";
    public static final String P_COMPARE_CLASS = "accumulo.filter.iterator.p.compare.class";
    public static final String COMPARE_OPT_CLASS = "accumulo.filter.iterator.compare.opt.class";
    public static final String CONST_VAL = "accumulo.filter.iterator.const.val";
    public static final String COLUMN = "accumulo.filter.iterator.qual";
    private String qual;
    private String cf;

    private CompareOp compOpt;

    private static final Logger log = Logger.getLogger(PrimitiveComparisonFilter.class);
    private static final Pattern PIPE_PATTERN = Pattern.compile("[|]");

    @Override
    protected boolean filter(Text currentRow, List<Key> keys, List<Value> values) {
        SortedMap<Key,Value> items;
        boolean allow;
        try {       //if key doesn't contain CF, it's an encoded value from a previous iterator.
            while(keys.get(0).getColumnFamily().getBytes().length == 0) {
                items = decodeRow(keys.get(0), values.get(0));
                keys = Lists.newArrayList(items.keySet());
// Path: src/main/java/org/apache/accumulo/storagehandler/predicate/compare/CompareOp.java
// public interface CompareOp {
//     /**
//      *
//      * @param comp
//      */
//     public void setPrimitiveCompare(PrimitiveCompare comp);
// 
//     /**
//      *
//      *
//      * @return PrimitiveCompare
//      */
//     public PrimitiveCompare getPrimitiveCompare();
// 
//     /**
//      *
//      *
//      * @param val
//      * @return boolean
//      */
//     public boolean accept(byte [] val);
// }
// 
// Path: src/main/java/org/apache/accumulo/storagehandler/predicate/compare/PrimitiveCompare.java
// public interface PrimitiveCompare {
// 
//     public boolean isEqual(byte[] value);
//     public boolean isNotEqual(byte[] value);
//     public boolean greaterThanOrEqual(byte[] value);
//     public boolean greaterThan(byte[] value);
//     public boolean lessThanOrEqual(byte[] value);
//     public boolean lessThan(byte[] value);
//     public boolean like(byte[] value);
//     public Object serialize(byte[] value);
//     public void init(byte[] constant);
// }


// Path: src/main/java/org/apache/accumulo/storagehandler/predicate/PrimitiveComparisonFilter.java
import com.google.common.collect.Lists;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.iterators.IteratorEnvironment;
import org.apache.accumulo.core.iterators.SortedKeyValueIterator;
import org.apache.accumulo.core.iterators.user.WholeRowIterator;
import org.apache.accumulo.storagehandler.predicate.compare.CompareOp;
import org.apache.accumulo.storagehandler.predicate.compare.PrimitiveCompare;
import org.apache.commons.codec.binary.Base64;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.*;
import java.util.regex.Pattern;

package org.apache.accumulo.storagehandler.predicate;



/**
 * Operates over a single qualifier.
 *
 * Delegates to PrimitiveCompare and CompareOpt instances for
 * value acceptance.
 *
 * The PrimitiveCompare strategy assumes a consistent value type for the same column family and qualifier.
 *
 */
public class PrimitiveComparisonFilter extends WholeRowIterator {

    public static final String FILTER_PREFIX = "accumulo.filter.compare.iterator.";
    public static final String P_COMPARE_CLASS = "accumulo.filter.iterator.p.compare.class";
    public static final String COMPARE_OPT_CLASS = "accumulo.filter.iterator.compare.opt.class";
    public static final String CONST_VAL = "accumulo.filter.iterator.const.val";
    public static final String COLUMN = "accumulo.filter.iterator.qual";
    private String qual;
    private String cf;

    private CompareOp compOpt;

    private static final Logger log = Logger.getLogger(PrimitiveComparisonFilter.class);
    private static final Pattern PIPE_PATTERN = Pattern.compile("[|]");

    @Override
    protected boolean filter(Text currentRow, List<Key> keys, List<Value> values) {
        SortedMap<Key,Value> items;
        boolean allow;
        try {       //if key doesn't contain CF, it's an encoded value from a previous iterator.
            while(keys.get(0).getColumnFamily().getBytes().length == 0) {
                items = decodeRow(keys.get(0), values.get(0));
                keys = Lists.newArrayList(items.keySet());
                values = Lists.newArrayList(items.values());