[16/55] [abbrv] [partial] incubator-eagle git commit: [EAGLE-46] Rename package name as "org.apache.eagle"

[ha...@apache.org]

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/HBaseFilterBuilder.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/HBaseFilterBuilder.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/HBaseFilterBuilder.java
new file mode 100755
index 0000000..8209445
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/HBaseFilterBuilder.java
@@ -0,0 +1,585 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.filter;
+
+import org.apache.eagle.common.config.EagleConfigFactory;
+import org.apache.eagle.log.entity.EntityQualifierUtils;
+import org.apache.eagle.log.entity.meta.EntityDefinition;
+import org.apache.eagle.log.entity.meta.Qualifier;
+import org.apache.eagle.common.ByteUtil;
+import org.apache.eagle.query.parser.*;
+import org.apache.hadoop.hbase.filter.*;
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+import org.apache.hadoop.hbase.filter.FilterList.Operator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.Charset;
+import java.util.*;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * the steps of building hbase filters
+ * 1. receive ORExpression from eagle-antlr
+ * 2. iterate all ANDExpression in ORExpression
+ *    2.1 put each ANDExpression to a new filter list with MUST_PASS_ONE option
+ *    2.2 iterate all AtomicExpression in ANDExpression
+ *       2.2.1 group AtomicExpression into 2 groups by looking up metadata, one is for tag filters, the other is for column filters
+ *       2.2.2 put the above 2 filters to a filter list with MUST_PASS_ALL option
+ */
+public class HBaseFilterBuilder {
+	private static final Logger LOG = LoggerFactory.getLogger(HBaseFilterBuilder.class);
+	
+	/**
+	 * syntax is @<fieldname>
+	 */
+//	private static final String fnRegex = "^@(.*)$";
+	private static final Pattern _fnPattern = TokenConstant.ID_PATTERN;// Pattern.compile(fnRegex);
+	private static final Charset _defaultCharset = Charset.forName("ISO-8859-1");
+
+	private ORExpression _orExpr;
+	private EntityDefinition _ed;
+	private boolean _filterIfMissing;
+	private Charset _charset = _defaultCharset;
+
+	/**
+	 * TODO: Verify performance impact
+	 *
+	 * @return
+	 */
+	public Set<String> getFilterFields() {
+		return _filterFields;
+	}
+
+	/**
+	 * Just add filter fields for expression filter
+	 */
+	private Set<String> _filterFields;
+
+	public HBaseFilterBuilder(EntityDefinition ed, ORExpression orExpr) {
+		this(ed, orExpr, false);
+	}
+	
+	public HBaseFilterBuilder(EntityDefinition ed, ORExpression orExpr, boolean filterIfMissing) {
+		this._ed = ed;
+		this._orExpr = orExpr;
+		this._filterIfMissing = filterIfMissing;
+	}
+	
+	public void setCharset(String charsetName){
+		_charset = Charset.forName(charsetName);
+	}
+	
+	public Charset getCharset(){
+		return _charset;
+	}
+	
+	/**
+	 * Because we don't have metadata for tag, we regard non-qualifer field as tag. So one field possibly is not a real tag when this function return true. This happens
+	 * when a user input an wrong field name which is neither tag or qualifier
+	 *   
+	 * @param field
+	 */
+	private boolean isTag(String field){
+		return _ed.isTag(field);
+	}
+	
+	/**
+	 * check whether this field is one entity attribute or not 
+	 * @param fieldName
+	 * @return
+	 */
+	private String parseEntityAttribute(String fieldName){
+		Matcher m = _fnPattern.matcher(fieldName);
+		if(m.find()){
+			return m.group(1);
+		}
+		return null;
+	}
+
+	/**
+	 * Return the partition values for each or expression. The size of the returned list should be equal to
+	 * the size of FilterList that {@link #buildFilters()} returns.
+	 * 
+	 * TODO: For now we don't support one query to query multiple partitions. In future if partition is defined, 
+	 * for the entity, internally We need to spawn multiple queries and send one query for each partition.
+	 * 
+	 * @return Return the partition values for each or expression. Return null if the entity doesn't support
+	 * partition
+	 */
+	public List<String[]> getPartitionValues() {
+		final String[] partitions = _ed.getPartitions();
+		if (partitions == null || partitions.length == 0) {
+			return null;
+		}
+		final List<String[]> result = new ArrayList<String[]>();
+		final Map<String, String> partitionKeyValueMap = new HashMap<String, String>();
+		for(ANDExpression andExpr : _orExpr.getANDExprList()) {
+			partitionKeyValueMap.clear();
+			for(AtomicExpression ae : andExpr.getAtomicExprList()) {
+				// TODO temporarily ignore those fields which are not for attributes
+				if(ae.getKeyType() == TokenType.ID) {
+					final String fieldName = parseEntityAttribute(ae.getKey());
+					if (fieldName == null) {
+						LOG.warn(fieldName + " field does not have format @<FieldName>, ignored");
+						continue;
+					}
+					if (_ed.isPartitionTag(fieldName) && ComparisonOperator.EQUAL.equals(ae.getOp())) {
+						final String value = ae.getValue();
+						partitionKeyValueMap.put(fieldName, value);
+					}
+				}
+			}
+			final String[] values = new String[partitions.length];
+			result.add(values);
+			for (int i = 0; i < partitions.length; ++i) {
+				final String partition = partitions[i];
+				final String value = partitionKeyValueMap.get(partition);
+				values[i] = value;
+			}
+		}
+		return result;
+	}
+
+	/**
+	 * @see org.apache.eagle.query.parser.TokenType
+	 *
+	 * @return
+	 */
+	public FilterList buildFilters(){
+		// TODO: Optimize to select between row filter or column filter for better performance
+		// Use row key filter priority by default
+		boolean rowFilterPriority = true;
+
+		FilterList fltList = new FilterList(Operator.MUST_PASS_ONE);
+		for(ANDExpression andExpr : _orExpr.getANDExprList()){
+			
+			FilterList list = new FilterList(Operator.MUST_PASS_ALL);
+			Map<String, List<String>> tagFilters = new HashMap<String, List<String>>();
+			List<QualifierFilterEntity> qualifierFilters = new ArrayList<QualifierFilterEntity>();
+//			List<QualifierFilterEntry> tagLikeQualifierFilters = new ArrayList<QualifierFilterEntry>();
+
+			// TODO refactor not to use too much if/else
+			for(AtomicExpression ae : andExpr.getAtomicExprList()){
+				// TODO temporarily ignore those fields which are not for attributes
+
+				String fieldName = ae.getKey();
+				if(ae.getKeyType() == TokenType.ID){
+					fieldName = parseEntityAttribute(fieldName);
+					if(fieldName == null){
+						LOG.warn(fieldName + " field does not have format @<FieldName>, ignored");
+						continue;
+					}
+				}
+
+				String value = ae.getValue();
+				ComparisonOperator op = ae.getOp();
+				TokenType keyType = ae.getKeyType();
+				TokenType valueType = ae.getValueType();
+				QualifierFilterEntity entry = new QualifierFilterEntity(fieldName,value,op,keyType,valueType);
+
+				// TODO Exact match, need to add escape for those special characters here, including:
+				// "-", "[", "]", "/", "{", "}", "(", ")", "*", "+", "?", ".", "\\", "^", "$", "|"
+
+				if(keyType == TokenType.ID && isTag(fieldName)){
+					if ((ComparisonOperator.EQUAL.equals(op) || ComparisonOperator.IS.equals(op))
+							&& !TokenType.NULL.equals(valueType))
+					{
+						// Use RowFilter for equal TAG
+						if(tagFilters.get(fieldName) == null) tagFilters.put(fieldName, new ArrayList<String>());
+						tagFilters.get(fieldName).add(value);
+					} else if (rowFilterPriority && ComparisonOperator.IN.equals(op))
+					{
+						// Use RowFilter here by default
+						if(tagFilters.get(fieldName) == null) tagFilters.put(fieldName, new ArrayList<String>());
+						tagFilters.get(fieldName).addAll(EntityQualifierUtils.parseList(value));
+					} else if (ComparisonOperator.LIKE.equals(op)
+						|| ComparisonOperator.NOT_LIKE.equals(op)
+						|| ComparisonOperator.CONTAINS.equals(op)
+						|| ComparisonOperator.NOT_CONTAINS.equals(op)
+						|| ComparisonOperator.IN.equals(op)
+						|| ComparisonOperator.IS.equals(op)
+						|| ComparisonOperator.IS_NOT.equals(op)
+						|| ComparisonOperator.NOT_EQUAL.equals(op)
+						|| ComparisonOperator.EQUAL.equals(op)
+						|| ComparisonOperator.NOT_IN.equals(op))
+					{
+						qualifierFilters.add(entry);
+					} else
+					{
+						LOG.warn("Don't support operation: \"" + op + "\" on tag field: " + fieldName + " yet, going to ignore");
+						throw new IllegalArgumentException("Don't support operation: "+op+" on tag field: "+fieldName+", avaliable options: =, =!, =~, !=~, in, not in, contains, not contains");
+					}
+				}else{
+					qualifierFilters.add(entry);
+				}
+			}
+
+			// Build RowFilter for equal tags
+			list.addFilter(buildTagFilter(tagFilters));
+
+			// Build SingleColumnValueFilter
+			FilterList qualifierFilterList = buildQualifierFilter(qualifierFilters);
+			if(qualifierFilterList != null && qualifierFilterList.getFilters().size()>0){
+				list.addFilter(qualifierFilterList);
+			}else {
+				if(LOG.isDebugEnabled()) LOG.debug("Ignore empty qualifier filter from "+qualifierFilters.toString());
+			}
+			fltList.addFilter(list);
+		}
+		LOG.info("Query: " + _orExpr.toString() + " => Filter: " + fltList.toString());
+		return fltList;
+	}
+	
+	/**
+	 * _charset is used to decode the byte array, in hbase server, RegexStringComparator uses the same
+	 * charset to decode the byte array stored in qualifier
+	 * for tag filter regex, it's always ISO-8859-1 as it only comes from String's hashcode (Integer)
+	 * Note: regex comparasion is to compare String
+	 */
+	protected Filter buildTagFilter(Map<String, List<String>> tagFilters){
+		RegexStringComparator regexStringComparator = new RegexStringComparator(buildTagFilterRegex(tagFilters));
+		regexStringComparator.setCharset(_charset);
+		RowFilter filter = new RowFilter(CompareOp.EQUAL, regexStringComparator);
+		return filter;
+	}
+	
+	/**
+	 * all qualifiers' condition must be satisfied.
+	 *
+	 * <H1>Use RegexStringComparator for:</H1>
+	 *      IN
+	 *      LIKE
+	 *      NOT_LIKE
+	 *
+	 * <H1>Use SubstringComparator for:</H1>
+	 *      CONTAINS
+	 *
+	 * <H1>Use EntityQualifierHelper for:</H1>
+	 *      EQUALS
+	 *      NOT_EUQALS
+	 *      LESS
+	 *      LESS_OR_EQUAL
+	 *      GREATER
+	 *      GREATER_OR_EQUAL
+	 *
+	 * <H2>
+	 *     TODO: Compare performance of RegexStringComparator ,SubstringComparator ,EntityQualifierHelper
+	 * </H2>
+	 *
+	 * @param qualifierFilters
+	 * @return
+	 */
+	protected FilterList buildQualifierFilter(List<QualifierFilterEntity> qualifierFilters){
+		FilterList list = new FilterList(Operator.MUST_PASS_ALL);
+		// iterate all the qualifiers
+		for(QualifierFilterEntity entry : qualifierFilters){
+			// if contains expression based filter
+			if(entry.getKeyType() == TokenType.EXP
+					|| entry.getValueType() == TokenType.EXP
+					|| entry.getKeyType() != TokenType.ID){
+				if(!EagleConfigFactory.load().isCoprocessorEnabled()) {
+					LOG.warn("Expression in filter may not support, because custom filter and coprocessor is disabled: " + entry.toString());
+				}
+				list.addFilter(buildExpressionBasedFilter(entry));
+				continue;
+			}
+
+			// else using SingleColumnValueFilter
+			String qualifierName = entry.getKey();
+			if(!isTag(entry.getKey())){
+				Qualifier qualifier = _ed.getDisplayNameMap().get(entry.getKey());
+				qualifierName = qualifier.getQualifierName();
+			}
+
+			// Comparator to be used for building HBase Filter
+			// WritableByteArrayComparable comparator;
+            ByteArrayComparable comparable;
+			if(ComparisonOperator.IN.equals(entry.getOp())
+				|| ComparisonOperator.NOT_IN.equals(entry.getOp())){
+				Filter setFilter = buildListQualifierFilter(entry);
+				if(setFilter!=null){
+					list.addFilter(setFilter);
+				}
+			}else{
+				// If [=,!=,is,is not] NULL, use NullComparator else throw exception
+				if(TokenType.NULL.equals(entry.getValueType())){
+					if(ComparisonOperator.EQUAL.equals(entry.getOp())
+						||ComparisonOperator.NOT_EQUAL.equals(entry.getOp())
+						||ComparisonOperator.IS.equals(entry.getOp())
+						||ComparisonOperator.IS_NOT.equals(entry.getOp()))
+                        comparable = new NullComparator();
+					else
+						throw new IllegalArgumentException("Operation: "+entry.getOp()+" with NULL is not supported yet: "+entry.toString()+", avaliable options: [=, !=, is, is not] null|NULL");
+				}
+				// If [contains, not contains],use SubstringComparator
+				else if (ComparisonOperator.CONTAINS.equals(entry.getOp())
+					|| ComparisonOperator.NOT_CONTAINS.equals(entry.getOp())) {
+                    comparable = new SubstringComparator(entry.getValue());
+				}
+				// If [like, not like], use RegexStringComparator
+				else if (ComparisonOperator.LIKE.equals(entry.getOp())
+						|| ComparisonOperator.NOT_LIKE.equals(entry.getOp())){
+					// Use RegexStringComparator for LIKE / NOT_LIKE
+					RegexStringComparator _comparator = new RegexStringComparator(buildQualifierRegex(entry.getValue()));
+					_comparator.setCharset(_charset);
+                    comparable = _comparator;
+				} else{
+					Class type = EntityQualifierUtils.getType(_ed, entry.getKey());
+					// if type is null (is Tag or not found) or not defined for TypedByteArrayComparator
+					if(!EagleConfigFactory.load().isCoprocessorEnabled() || type == null || TypedByteArrayComparator.get(type) == null){
+                        comparable = new BinaryComparator(EntityQualifierUtils.toBytes(_ed, entry.getKey(), entry.getValue()));
+					}else {
+                        comparable = new TypedByteArrayComparator(EntityQualifierUtils.toBytes(_ed, entry.getKey(), entry.getValue()),type);
+					}
+				}
+
+				SingleColumnValueFilter filter =
+						new SingleColumnValueFilter(_ed.getColumnFamily().getBytes(), qualifierName.getBytes(), convertToHBaseCompareOp(entry.getOp()), comparable);
+				filter.setFilterIfMissing(_filterIfMissing);
+				list.addFilter(filter);
+			}
+		}
+
+		return list;
+	}
+
+	private Filter buildExpressionBasedFilter(QualifierFilterEntity entry) {
+		BooleanExpressionComparator expressionComparator  = new BooleanExpressionComparator(entry,_ed);
+		_filterFields = expressionComparator.getRequiredFields();
+		RowValueFilter filter = new RowValueFilter(expressionComparator);
+		return filter;
+	}
+
+	/**
+	 * Currently use BinaryComparator only
+	 * <h2>TODO: </h2>
+	 * Possibility to tune performance by using: OR[BinaryComparator,...] instead of RegexStringComparator?
+	 *
+	 *<br/> <br/>
+	 *
+	 * ! Check op must be IN or NOTIN in caller
+	 *
+	 * @param entry
+	 * @return
+	 */
+	private Filter buildListQualifierFilter(QualifierFilterEntity entry){
+		List<String> valueSet = EntityQualifierUtils.parseList(entry.getValue());
+		Iterator<String> it = valueSet.iterator();
+		String fieldName = entry.getKey();
+		String qualifierName = fieldName;
+		if(!_ed.isTag(entry.getKey())){
+			qualifierName = _ed.getDisplayNameMap().get(entry.getKey()).getQualifierName();
+		}
+
+// TODO: Try to use RegExp just work if possible
+// Because single SingleColumnValueFilter is much faster than multi SingleColumnValueFilters in OR list.
+//		Class qualifierType = EntityQualifierHelper.getType(_ed,fieldName);
+//		if( qualifierType == null || qualifierType == String.class){
+//			boolean first = true;
+//			StringBuilder filterRegex = new StringBuilder();
+//			filterRegex.append("^(");
+//			while(it.hasNext()) {
+//				String value = it.next();
+//				if(value == null) {
+//					logger.warn("ignore empty value in set qualifier filter: "+entry.toString());
+//					continue;
+//				}
+//				if(!first) filterRegex.append("|");
+//				filterRegex.append(value);
+//				first = false;
+//			}
+//			filterRegex.append(")$");
+//			RegexStringComparator regexStringComparator = new RegexStringComparator(filterRegex.toString());
+//			return new SingleColumnValueFilter(_ed.getColumnFamily().getBytes(), qualifierName.getBytes(),
+//					convertToHBaseCompareOp(entry.getOp()), regexStringComparator);
+//		}else{
+		FilterList setFilterList;
+		if(ComparisonOperator.IN.equals(entry.getOp())){
+			setFilterList = new FilterList(Operator.MUST_PASS_ONE);
+		}else if(ComparisonOperator.NOT_IN.equals(entry.getOp())) {
+			setFilterList = new FilterList(Operator.MUST_PASS_ALL);
+		}else{
+			throw new IllegalArgumentException(String.format("Don't support operation: %s on LIST type of value yet: %s, valid options: IN/NOT IN [LIST]",entry.getOp(),entry.toString()));
+		}
+
+		while(it.hasNext()) {
+			String value = it.next();
+			BinaryComparator comparator = new BinaryComparator(EntityQualifierUtils.toBytes(_ed, fieldName, value));
+			SingleColumnValueFilter filter =
+					new SingleColumnValueFilter(_ed.getColumnFamily().getBytes(), qualifierName.getBytes(), convertToHBaseCompareOp(entry.getOp()), comparator);
+			filter.setFilterIfMissing(_filterIfMissing);
+			setFilterList.addFilter(filter);
+		}
+
+		return setFilterList;
+//		}
+	}
+
+	/**
+	 * Just used for LIKE and NOT_LIKE
+	 *
+	 * @param qualifierValue
+	 * @return
+	 */
+	protected String buildQualifierRegex(String qualifierValue){
+		StringBuilder sb = new StringBuilder();
+//		sb.append("(?s)");
+		sb.append("^");
+		sb.append(qualifierValue);
+		sb.append("$");
+		return sb.toString();
+	}
+	
+	  /**
+	   * Appends the given ID to the given buffer, followed by "\\E".
+	   * [steal it from opentsdb, thanks opentsdb :) https://github.com/OpenTSDB/opentsdb/blob/master/src/core/TsdbQuery.java]
+	   */
+	  private static void addId(final StringBuilder buf, final byte[] id) {
+		buf.append("\\Q");
+	    boolean backslash = false;
+	    for (final byte b : id) {
+	      buf.append((char) (b & 0xFF));
+	      if (b == 'E' && backslash) {  // If we saw a `\' and now we have a `E'.
+	        // So we just terminated the quoted section because we just added \E
+	        // to `buf'.  So let's put a litteral \E now and start quoting again.
+	        buf.append("\\\\E\\Q");
+	      } else {
+	        backslash = b == '\\';
+	      }
+	    }
+	    buf.append("\\E");
+	  }
+	  
+	  @SuppressWarnings("unused")
+	  private static void addId(final StringBuilder buf, final String id) {
+		    buf.append("\\Q");
+		  	int len = id.length()-1;
+		    boolean backslash = false;
+		    for (int i =0; i < len; i++) {
+		      char c = id.charAt(i);
+		      buf.append(c);
+		      if (c == 'E' && backslash) {  // If we saw a `\' and now we have a `E'.
+		        // So we just terminated the quoted section because we just added \E
+		        // to `buf'.  So let's put a litteral \E now and start quoting again.
+		        buf.append("\\\\E\\Q");
+		      } else {
+		        backslash = c == '\\';
+		      }
+		    }
+		    buf.append("\\E");
+		  }
+	
+	/**
+	 * one search tag may have multiple values which have OR relationship, and relationship between
+	 * different search tags is AND
+	 * the query is like "(TAG1=value11 OR TAG1=value12) AND TAG2=value2"
+	 * @param tags
+	 * @return
+	 */
+	protected String buildTagFilterRegex(Map<String, List<String>> tags){
+		// TODO need consider that \E could be part of tag, refer to https://github.com/OpenTSDB/opentsdb/blob/master/src/core/TsdbQuery.java
+		final SortedMap<Integer, List<Integer>> tagHash = new TreeMap<Integer, List<Integer>>();
+		final int numOfPartitionFields = (_ed.getPartitions() == null) ? 0 : _ed.getPartitions().length;
+		for(Map.Entry<String, List<String>> entry : tags.entrySet()){
+			String tagName = entry.getKey();
+			// Ignore tag if the tag is one of partition fields
+			if (_ed.isPartitionTag(tagName)) {
+				continue;
+			}
+			List<String> stringValues = entry.getValue();
+			List<Integer> hashValues = new ArrayList<Integer>(stringValues.size());
+			for(String value : stringValues){
+				hashValues.add(value.hashCode());
+			}
+			tagHash.put(tagName.hashCode(), hashValues);
+		}
+		
+		// header = prefix(4 bytes) + partition_hashes(4*N bytes) + timestamp (8 bytes)
+		final int headerLength = 4 + numOfPartitionFields * 4 + 8;
+
+		// <tag1:4><value1:4> ... <tagn:4><valuen:4>
+		StringBuilder sb = new StringBuilder();
+		sb.append("(?s)");
+		sb.append("^(?:.{").append(headerLength).append("})");
+		sb.append("(?:.{").append(8).append("})*"); // for any number of tags
+		for (Map.Entry<Integer, List<Integer>> entry : tagHash.entrySet()) {
+			try {
+				addId(sb, ByteUtil.intToBytes(entry.getKey()));
+				List<Integer> hashValues = entry.getValue();
+				sb.append("(?:");
+				boolean first = true;
+				for(Integer value : hashValues){
+					if(!first){
+						sb.append('|');
+					}
+					addId(sb, ByteUtil.intToBytes(value));
+					first = false;
+				}
+				sb.append(")");
+				sb.append("(?:.{").append(8).append("})*"); // for any number of tags
+			} catch (Exception ex) {
+				LOG.error("constructing regex error", ex);
+			}
+		}
+		sb.append("$");
+		if(LOG.isDebugEnabled()) LOG.debug("Tag filter pattern is " + sb.toString());
+		return sb.toString();
+	}
+
+	/**
+	 * Convert ComparisonOperator to native HBase CompareOp
+	 *
+	 * Support:
+	 *      =, =~,CONTAINS,<,<=,>,>=,!=,!=~
+	 *
+	 * @param comp
+	 * @return
+	 */
+	protected static CompareOp convertToHBaseCompareOp(ComparisonOperator comp) {
+		if(comp == ComparisonOperator.EQUAL || comp == ComparisonOperator.LIKE
+				|| comp == ComparisonOperator.CONTAINS
+				|| comp == ComparisonOperator.IN
+				|| comp == ComparisonOperator.IS
+				) {
+			return CompareOp.EQUAL;
+		}else if(comp == ComparisonOperator.LESS) {
+			return CompareOp.LESS;
+		} else if(comp == ComparisonOperator.LESS_OR_EQUAL){
+			return CompareOp.LESS_OR_EQUAL;
+		}else if(comp == ComparisonOperator.GREATER) {
+			return CompareOp.GREATER;
+		} else if(comp == ComparisonOperator.GREATER_OR_EQUAL){
+			return CompareOp.GREATER_OR_EQUAL;
+		} else if(comp == ComparisonOperator.NOT_EQUAL
+				|| comp == ComparisonOperator.NOT_LIKE
+				|| comp == ComparisonOperator.NOT_CONTAINS
+				|| comp == ComparisonOperator.IS_NOT
+				|| comp == ComparisonOperator.NOT_IN)
+		{
+			return CompareOp.NOT_EQUAL;
+		} else {
+			LOG.error("{} operation is not supported now\n", comp);
+			throw new IllegalArgumentException("Illegal operation: "+comp+ ", avaliable options: "+ Arrays.toString(ComparisonOperator.values()));
+		}
+	}
+
+	protected static CompareOp getHBaseCompareOp(String comp) {
+		return convertToHBaseCompareOp(ComparisonOperator.locateOperator(comp));
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/QualifierFilterEntity.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/QualifierFilterEntity.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/QualifierFilterEntity.java
new file mode 100755
index 0000000..6cdc77b
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/QualifierFilterEntity.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.filter;
+
+import org.apache.eagle.query.parser.ComparisonOperator;
+import org.apache.eagle.query.parser.TokenType;
+import org.apache.hadoop.io.Writable;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+public class QualifierFilterEntity implements Writable{
+	public String key;
+	public String value;
+	public ComparisonOperator op;
+	public TokenType valueType;
+	public TokenType keyType;
+
+	public QualifierFilterEntity(){}
+	public QualifierFilterEntity(String key, String value, ComparisonOperator comp, TokenType keyType, TokenType valueType) {
+		super();
+		this.key = key;
+		this.value = value;
+		this.op = comp;
+		this.keyType = keyType;
+		this.valueType = valueType;
+	}
+
+	public String getKey() {
+		return key;
+	}
+
+	public void setKey(String key) {
+		this.key = key;
+	}
+
+	public String getValue() {
+		return value;
+	}
+
+	public void setValue(String value) {
+		this.value = value;
+	}
+
+	public ComparisonOperator getOp() {
+		return op;
+	}
+
+	public void setOp(ComparisonOperator op) {
+		this.op = op;
+	}
+
+	public TokenType getValueType() {
+		return valueType;
+	}
+
+	public void setValueType(TokenType valueType) {
+		this.valueType = valueType;
+	}
+
+	public void setKeyType(TokenType keyType){
+		this.keyType = keyType;
+	}
+	public TokenType getKeyType(){
+		return this.keyType;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("%s %s %s",this.key,this.op,this.value);
+	}
+
+	@Override
+	public void write(DataOutput out) throws IOException {
+		out.writeUTF(this.key);
+		out.writeUTF(this.getValue());
+		out.writeUTF(this.op.name());
+		out.writeUTF(this.keyType.name());
+		out.writeUTF(this.valueType.name());
+	}
+
+	@Override
+	public void readFields(DataInput in) throws IOException {
+		this.key = in.readUTF();
+		this.value = in.readUTF();
+		this.op = ComparisonOperator.valueOf(in.readUTF());
+		this.keyType = TokenType.valueOf(in.readUTF());
+		this.valueType = TokenType.valueOf(in.readUTF());
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/RowValueFilter.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/RowValueFilter.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/RowValueFilter.java
new file mode 100755
index 0000000..a4b97ea
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/RowValueFilter.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.filter;
+
+import com.google.common.io.ByteArrayDataInput;
+import com.google.common.io.ByteArrayDataOutput;
+import com.google.common.io.ByteStreams;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterBase;
+import org.apache.hadoop.io.WritableComparable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * TODO: Critical performance problem!!!
+ * TODO: Refactor to specified multi-column filter so that avoid return all qualifier columns from region server to client side
+ *
+ * @since 2014/11/17
+ */
+public class RowValueFilter extends FilterBase {
+    private final static Logger LOG = LoggerFactory.getLogger(RowValueFilter.class);
+    private boolean filterOutRow = false;
+    private WritableComparable<List<KeyValue>> comparator;
+
+    // TODO: Use qualifiers to reduce network tranfer
+//    private List<byte[]> qualifiers;
+    public RowValueFilter(){}
+
+    /**
+     * Filter out row if WritableComparable.compareTo return 0
+     * @param comparator <code>WritableComparable[List[KeyValue]]</code>
+     */
+    public RowValueFilter(WritableComparable<List<KeyValue>> comparator){
+        this.comparator = comparator;
+    }
+
+//    public RowValueFilter(List<byte[]> qualifiers,WritableComparable<List<KeyValue>> comparator){
+//        this.qualifiers = qualifiers;
+//        this.comparator = comparator;
+//    }
+
+    /**
+     * Old interface in hbase-0.94
+     *
+     * @param out
+     * @throws IOException
+     */
+    @Deprecated
+    public void write(DataOutput out) throws IOException {
+        this.comparator.write(out);
+    }
+
+    /**
+     * Old interface in hbase-0.94
+     *
+     * @param in
+     * @throws IOException
+     */
+//    @Override
+    @Deprecated
+    public void readFields(DataInput in) throws IOException {
+        this.comparator = new BooleanExpressionComparator();
+        this.comparator.readFields(in);
+    }
+
+    /**
+     * TODO: Currently still use older serialization method from hbase-0.94, need to migrate into ProtoBuff based
+     *
+     * @return
+     * @throws IOException
+     */
+    @Override
+    public byte[] toByteArray() throws IOException {
+        ByteArrayDataOutput byteArrayDataOutput = ByteStreams.newDataOutput();
+        this.comparator.write(byteArrayDataOutput);
+        return byteArrayDataOutput.toByteArray();
+    }
+
+    /**
+     * TODO: Currently still use older serialization method from hbase-0.94, need to migrate into ProtoBuff based
+     */
+    // Override static method
+    public static Filter parseFrom(final byte [] pbBytes) throws DeserializationException {
+        ByteArrayDataInput byteArrayDataInput = ByteStreams.newDataInput(pbBytes);
+        RowValueFilter filter = new RowValueFilter();
+        try {
+            filter.readFields(byteArrayDataInput);
+        } catch (IOException e) {
+            LOG.error("Got error to deserialize RowValueFilter from PB bytes",e);
+            throw new DeserializationException(e);
+        }
+        return filter;
+    }
+
+    @Override
+    public boolean hasFilterRow(){
+        return true;
+    }
+
+    @Override
+    public void filterRow(List<KeyValue> row) {
+        filterOutRow = (this.comparator.compareTo(row) == 0);
+    }
+
+    @Override
+    public void reset() {
+        this.filterOutRow = false;
+    }
+
+    @Override
+    public boolean filterRow(){
+        return filterOutRow;
+    }
+
+    @Override
+    public String toString() {
+        return super.toString()+" ( "+this.comparator.toString()+" )";
+    }
+
+//    public List<byte[]> getQualifiers() {
+//        return qualifiers;
+//    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/TypedByteArrayComparator.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/TypedByteArrayComparator.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/TypedByteArrayComparator.java
new file mode 100755
index 0000000..ecaf8cc
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/filter/TypedByteArrayComparator.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.filter;
+
+import com.google.common.io.ByteArrayDataInput;
+import com.google.common.io.ByteArrayDataOutput;
+import com.google.common.io.ByteStreams;
+import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.filter.ByteArrayComparable;
+import org.apache.hadoop.io.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * <h1>TypedByteArrayComparator</h1>
+ *
+ * Compare byte array: <code>byte[] value</code> with class type: <code>Class type</code>
+ *
+ * <br/>
+ * <br/>
+ * Built-in support:
+ *
+ *  <pre>
+ *    Double
+ *    double
+ *    Integer
+ *    int
+ *    Long
+ *    long
+ *    Short
+ *    short
+ *    Boolean
+ *    boolean
+ *  </pre>
+ *
+ *  And can be extend by defining new {@link RawComparator} and register with  {@link #define(Class type, RawComparator comparator)}
+ * <br/>
+ * <br/>
+ */
+public class TypedByteArrayComparator extends ByteArrayComparable {
+    private final static Logger LOG = LoggerFactory.getLogger(TypedByteArrayComparator.class);
+
+    private Class type;
+
+    // Not need to be writable
+    private RawComparator comparator;
+
+    /**
+     * Default constructor for writable
+     */
+    @SuppressWarnings("unused")
+    public TypedByteArrayComparator(){
+        super(null);
+    }
+
+    public TypedByteArrayComparator(byte[] value, Class type){
+        super(value);
+        this.type = type;
+        this.comparator = get(this.type);
+        if(this.comparator == null) throw new IllegalArgumentException("No comparator found for class: "+type);
+    }
+
+    /**
+     * @param in hbase-0.94 interface
+     * @throws IOException
+     */
+//    @Override
+    public void readFields(DataInput in) throws IOException {
+//        super.readFields(in);
+        try {
+            String _type = in.readUTF();
+            type = _primitiveTypeClassMap.get(_type);
+            if(type == null) {
+                type = Class.forName(_type);
+            }
+            comparator = get(type);
+            if(comparator == null) throw new IllegalArgumentException("No comparator found for class: "+type);
+        } catch (ClassNotFoundException e) {
+            throw new IOException(e.getMessage(),e);
+        }
+    }
+
+    /**
+     * @param out hbase-0.94 interface
+     * @throws IOException
+     */
+//    @Override
+    public void write(DataOutput out) throws IOException {
+//        super.write(out);
+        String typeName = type.getName();
+        out.writeUTF(typeName);
+    }
+
+    /**
+     * For hbase 0.98
+     *
+     * @return serialized byte array
+     */
+    @Override
+    public byte[] toByteArray() {
+        ByteArrayDataOutput byteArrayDataOutput = ByteStreams.newDataOutput();
+        try {
+            this.write(byteArrayDataOutput);
+            return byteArrayDataOutput.toByteArray();
+        } catch (IOException e) {
+            LOG.error("Failed to serialize due to: "+e.getMessage(),e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * For hbase 0.98
+     *
+     * @param bytes raw byte array
+     * @return Comparator instance
+     * @throws DeserializationException
+     */
+    public static TypedByteArrayComparator parseFrom(final byte [] bytes)
+            throws DeserializationException {
+        TypedByteArrayComparator comparator = new TypedByteArrayComparator();
+        ByteArrayDataInput byteArrayDataInput = ByteStreams.newDataInput(bytes);
+        try {
+            comparator.readFields(byteArrayDataInput);
+        } catch (IOException e) {
+            LOG.error("Got error to deserialize TypedByteArrayComparator from PB bytes",e);
+            throw new DeserializationException(e);
+        }
+        return comparator;
+    }
+
+    @Override
+    public int compareTo(byte[] value, int offset, int length) {
+        return this.comparator.compare(this.getValue(), 0, this.getValue().length, value, offset, length);
+    }
+
+    /**
+     * <ol>
+     * <li>Try registered comparator</li>
+     * <li>If not found, try all possible WritableComparator</li>
+     * </ol>
+     *
+     * If not found finally, throw new IllegalArgumentException("unable to get comparator for class: "+type);
+     *
+     * @param type value type class
+     * @return RawComparator
+     */
+    public static RawComparator get(Class type){
+        RawComparator comparator = null;
+        try {
+            comparator = _typedClassComparator.get(type);
+        }catch (ClassCastException ex){
+            // ignore
+        }
+        try {
+            if (comparator == null) comparator = WritableComparator.get(type);
+        }catch (ClassCastException ex){
+            // ignore
+        }
+        return comparator;
+    }
+
+    private final static Map<Class,RawComparator> _typedClassComparator = new HashMap<Class, RawComparator>();
+    public static void define(Class type, RawComparator comparator){
+        _typedClassComparator.put(type,comparator);
+    }
+
+    static{
+        define(Double.class, WritableComparator.get(DoubleWritable.class));
+        define(double.class, WritableComparator.get(DoubleWritable.class));
+        define(Integer.class, WritableComparator.get(IntWritable.class));
+        define(int.class, WritableComparator.get(IntWritable.class));
+        define(Long.class, WritableComparator.get(LongWritable.class));
+        define(long.class, WritableComparator.get(LongWritable.class));
+        define(Short.class, WritableComparator.get(ShortWritable.class));
+        define(short.class, WritableComparator.get(ShortWritable.class));
+        define(Boolean.class, WritableComparator.get(BooleanWritable.class));
+        define(boolean.class, WritableComparator.get(BooleanWritable.class));
+    }
+
+    /**
+     * Because {@link Class#forName } can't find class for primitive type
+     */
+    private final static Map<String,Class> _primitiveTypeClassMap = new HashMap<String, Class>();
+    static {
+        _primitiveTypeClassMap.put(int.class.getName(),int.class);
+        _primitiveTypeClassMap.put(double.class.getName(),double.class);
+        _primitiveTypeClassMap.put(long.class.getName(),long.class);
+        _primitiveTypeClassMap.put(short.class.getName(),short.class);
+        _primitiveTypeClassMap.put(boolean.class.getName(),boolean.class);
+        _primitiveTypeClassMap.put(char.class.getName(),char.class);
+        _primitiveTypeClassMap.put(byte.class.getName(),byte.class);
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexLogReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexLogReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexLogReader.java
new file mode 100755
index 0000000..418ab33
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexLogReader.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import org.apache.eagle.log.entity.LogReader;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.filter.Filter;
+import org.apache.hadoop.hbase.filter.FilterList;
+import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
+
+public abstract class IndexLogReader implements LogReader {
+
+	// TODO: Work around https://issues.apache.org/jira/browse/HBASE-2198. More graceful implementation should use SingleColumnValueExcludeFilter, 
+	// but it's complicated in current implementation. 
+	protected static void workaroundHBASE2198(Get get, Filter filter,byte[][] qualifiers) {
+		if (filter instanceof SingleColumnValueFilter) {
+			if(qualifiers == null) {
+				get.addFamily(((SingleColumnValueFilter) filter).getFamily());
+			}else{
+				get.addColumn(((SingleColumnValueFilter) filter).getFamily(), ((SingleColumnValueFilter) filter).getQualifier());
+			}
+			return;
+		}
+		if (filter instanceof FilterList) {
+			for (Filter f : ((FilterList)filter).getFilters()) {
+				workaroundHBASE2198(get, f,qualifiers);
+			}
+		}
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexStreamReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexStreamReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexStreamReader.java
new file mode 100755
index 0000000..9e059f2
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/IndexStreamReader.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import org.apache.eagle.log.base.taggedlog.TaggedLogAPIEntity;
+import org.apache.eagle.log.entity.*;
+import org.apache.eagle.log.entity.meta.EntityDefinition;
+import org.apache.eagle.log.entity.meta.IndexDefinition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+public abstract class IndexStreamReader  extends StreamReader {
+	protected final IndexDefinition indexDef;
+	protected final SearchCondition condition;
+	protected final List<byte[]> indexRowkeys;
+	protected LogReader<InternalLog> reader;
+	protected long lastTimestamp = 0;
+	protected long firstTimestamp = 0;
+	
+	protected static final Logger LOG = LoggerFactory.getLogger(IndexStreamReader.class);
+
+	public IndexStreamReader(IndexDefinition indexDef, SearchCondition condition, List<byte[]> indexRowkeys) {
+		this.indexDef = indexDef;
+		this.condition = condition;
+		this.indexRowkeys = indexRowkeys;
+		this.reader = null;
+	}
+
+	@Override
+	public long getLastTimestamp() {
+		return lastTimestamp;
+	}
+
+	@Override
+	public long getFirstTimestamp() {
+		return this.firstTimestamp;
+	}
+
+	@Override
+	public void readAsStream() throws Exception {
+		if (reader == null) {
+			reader = createIndexReader();
+		}
+		final EntityDefinition entityDef = indexDef.getEntityDefinition();
+		try{
+			reader.open();
+			InternalLog log;
+			int count = 0;
+			while ((log = reader.read()) != null) {
+				TaggedLogAPIEntity entity = HBaseInternalLogHelper.buildEntity(log, entityDef);
+				entity.setSerializeAlias(condition.getOutputAlias());
+				entity.setSerializeVerbose(condition.isOutputVerbose());
+
+				if (lastTimestamp == 0 || lastTimestamp < entity.getTimestamp()) {
+					lastTimestamp = entity.getTimestamp();
+				}
+				if(firstTimestamp == 0 || firstTimestamp > entity.getTimestamp()){
+					firstTimestamp = entity.getTimestamp();
+				}
+				for(EntityCreationListener l : _listeners){
+					l.entityCreated(entity);
+				}
+				if(++count == condition.getPageSize()) {
+					break;
+				}
+			}
+		}catch(IOException ioe){
+			LOG.error("Fail reading log", ioe);
+			throw ioe;
+		}finally{
+			reader.close();
+		}		
+	}
+
+	protected abstract LogReader createIndexReader();
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexLogReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexLogReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexLogReader.java
new file mode 100755
index 0000000..e6a5c96
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexLogReader.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import org.apache.eagle.common.config.EagleConfigFactory;
+import org.apache.eagle.log.entity.HBaseInternalLogHelper;
+import org.apache.eagle.log.entity.InternalLog;
+import org.apache.eagle.log.entity.meta.IndexDefinition;
+import org.apache.eagle.common.ByteUtil;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.filter.Filter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class NonClusteredIndexLogReader extends IndexLogReader {
+	private final IndexDefinition indexDef;
+	private final List<byte[]> indexRowkeys;
+	private final byte[][] qualifiers;
+	private final Filter filter;
+	private HTableInterface tbl;
+	private boolean isOpen = false;
+	private Result[] results;
+	private int index = -1;
+	private final List<Scan> scans;
+	private int currentScanIndex = 0;
+	private ResultScanner currentResultScanner;
+
+	// Max tag key/value. 
+	private static final byte[] MAX_TAG_VALUE_BYTES = {(byte) 0XFF,(byte) 0XFF,(byte) 0XFF,(byte) 0XFF,(byte) 0XFF,(byte) 0XFF,(byte) 0XFF,(byte) 0XFF,(byte) 0XFF};
+	private static final int BATCH_MULTIGET_SIZE = 1000;
+
+	public NonClusteredIndexLogReader(IndexDefinition indexDef, List<byte[]> indexRowkeys, byte[][] qualifiers, Filter filter) {
+		this.indexDef = indexDef;
+		this.indexRowkeys = indexRowkeys;
+		this.qualifiers = qualifiers;
+		this.filter = filter;
+		this.scans = buildScans();
+	}
+	
+
+	private List<Scan> buildScans() {
+		final ArrayList<Scan> result = new ArrayList<Scan>(indexRowkeys.size());
+		for (byte[] rowkey : indexRowkeys) {
+			Scan s = new Scan();
+			s.setStartRow(rowkey);
+			// In rowkey the tag key/value is sorted by the hash code of the key, so MAX_TAG_VALUE_BYTES is enough as the end key
+			final byte[] stopRowkey = ByteUtil.concat(rowkey, MAX_TAG_VALUE_BYTES);
+			s.setStopRow(stopRowkey);
+			// TODO the # of cached rows should be minimum of (pagesize and 100)
+			int cs = EagleConfigFactory.load().getHBaseClientScanCacheSize();
+			s.setCaching(cs);
+			// TODO not optimized for all applications
+			s.setCacheBlocks(true);
+			// scan specified columnfamily for all qualifiers
+			s.addFamily(indexDef.getEntityDefinition().getColumnFamily().getBytes());
+			result.add(s);
+		}
+		return result;
+	}
+
+	@Override
+	public void open() throws IOException {
+		if (isOpen)
+			return; // silently return
+		try {
+			tbl = EagleConfigFactory.load().getHTable(indexDef.getEntityDefinition().getTable());
+		} catch (RuntimeException ex) {
+			throw new IOException(ex);
+		}
+		currentScanIndex = 0;
+		openNewScan();
+		fillResults();
+	}
+
+	private boolean openNewScan() throws IOException {
+		closeCurrentScanResult();
+		if (currentScanIndex >= scans.size()) {
+			return false;
+		}
+		final Scan scan = scans.get(currentScanIndex++);
+		currentResultScanner = tbl.getScanner(scan);
+		return true;
+	}
+
+	private void fillResults() throws IOException {
+		if (currentResultScanner == null) {
+			return;
+		}
+		index = 0;
+		int count = 0;
+		Result r = null;
+        final List<Get> gets = new ArrayList<Get>(BATCH_MULTIGET_SIZE);
+		final byte[] family = indexDef.getEntityDefinition().getColumnFamily().getBytes();
+		while (count < BATCH_MULTIGET_SIZE) {
+			r = currentResultScanner.next();
+			if (r == null) {
+				if (openNewScan()) {
+					continue;
+				} else {
+					break;
+				}
+			}
+			for (byte[] rowkey : r.getFamilyMap(family).keySet()) {
+				if (rowkey.length == 0) {	// invalid rowkey
+					continue;
+				}
+				final Get get = new Get(rowkey);
+                if (filter != null) {
+                	get.setFilter(filter);
+                }
+				if(qualifiers != null) {
+					for (int j = 0; j < qualifiers.length; ++j) {
+						// Return the specified qualifiers
+						get.addColumn(family, qualifiers[j]);
+					}
+				}else {
+					get.addFamily(family);
+				}
+        		workaroundHBASE2198(get, filter,qualifiers);
+				gets.add(get);
+				++count;
+			}
+		}
+		if (count == 0) {
+			results = null;
+			return;
+		}
+		results = tbl.get(gets);
+		if (results == null || results.length == 0) {
+			fillResults();
+		}
+	}
+
+
+	private void closeCurrentScanResult() {
+		if (currentResultScanner != null) {
+			currentResultScanner.close();
+			currentResultScanner = null;
+		}
+	}
+
+
+	@Override
+	public void close() throws IOException {
+		if(tbl != null){
+			new HTableFactory().releaseHTableInterface(tbl);
+		}
+		closeCurrentScanResult();
+	}
+
+	@Override
+	public InternalLog read() throws IOException {
+		if (tbl == null) {
+			throw new IllegalArgumentException("Haven't open before reading");
+		}
+		
+		Result r = null;
+		InternalLog t = null;
+		while ((r = getNextResult()) != null) {
+			if (r.getRow() == null) {
+				continue;
+			}
+			t = HBaseInternalLogHelper.parse(indexDef.getEntityDefinition(), r, qualifiers);
+			break;
+		}
+		return t;
+	}
+
+
+	private Result getNextResult() throws IOException {
+		if (results == null || results.length == 0 || index >= results.length) {
+			fillResults();
+		}
+		if (results == null || results.length == 0 || index >= results.length) {
+			return null;
+		}
+		return results[index++];
+	}
+	
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexStreamReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexStreamReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexStreamReader.java
new file mode 100755
index 0000000..ec5631a
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/NonClusteredIndexStreamReader.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import org.apache.eagle.log.entity.HBaseInternalLogHelper;
+import org.apache.eagle.log.entity.LogReader;
+import org.apache.eagle.log.entity.SearchCondition;
+import org.apache.eagle.log.entity.meta.EntityDefinition;
+import org.apache.eagle.log.entity.meta.IndexDefinition;
+import org.apache.eagle.log.entity.meta.IndexDefinition.IndexType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class NonClusteredIndexStreamReader extends IndexStreamReader {
+	public NonClusteredIndexStreamReader(IndexDefinition indexDef, SearchCondition condition) {
+		super(indexDef, condition, new ArrayList<byte[]>());
+		final IndexType type = indexDef.canGoThroughIndex(condition.getQueryExpression(), indexRowkeys);
+		if (!IndexType.NON_CLUSTER_INDEX.equals(type)) {
+			throw new IllegalArgumentException("This query can't go through index: " + condition.getQueryExpression());
+		}
+	}
+
+	public NonClusteredIndexStreamReader(IndexDefinition indexDef, SearchCondition condition, List<byte[]> indexRowkeys) {
+		super(indexDef, condition, indexRowkeys);
+	}
+
+	@Override
+	protected LogReader createIndexReader() {
+		final EntityDefinition entityDef = indexDef.getEntityDefinition();
+		byte[][] outputQualifiers = null;
+		if(!condition.isOutputAll()) {
+			outputQualifiers = HBaseInternalLogHelper.getOutputQualifiers(entityDef, condition.getOutputFields());
+		}
+		return new NonClusteredIndexLogReader(indexDef, indexRowkeys, outputQualifiers, condition.getFilter());
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java
new file mode 100755
index 0000000..1c16dc8
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/RowKeyLogReader.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.eagle.common.config.EagleConfigFactory;
+import org.apache.eagle.log.entity.HBaseInternalLogHelper;
+import org.apache.eagle.log.entity.InternalLog;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTableFactory;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Result;
+
+import org.apache.eagle.log.entity.meta.EntityDefinition;
+
+public class RowKeyLogReader extends IndexLogReader {
+	private final EntityDefinition ed;
+	private final List<byte[]> rowkeys;
+    private final byte[][] qualifiers;
+    private HTableInterface tbl;
+	private boolean isOpen = false;
+	private Result[] entityResult;
+    private int getIndex = -1;
+
+    public RowKeyLogReader(EntityDefinition ed, byte[] rowkey) {
+        this.ed = ed;
+        this.rowkeys = new ArrayList<>();
+        this.rowkeys.add(rowkey);
+        this.qualifiers = null;
+    }
+
+	public RowKeyLogReader(EntityDefinition ed, byte[] rowkey,byte[][] qualifiers) {
+		this.ed = ed;
+		this.rowkeys = new ArrayList<>();
+        this.rowkeys.add(rowkey);
+        this.qualifiers = qualifiers;
+	}
+
+	public RowKeyLogReader(EntityDefinition ed, List<byte[]> rowkeys,byte[][] qualifiers) {
+		this.ed = ed;
+		this.rowkeys = rowkeys;
+        this.qualifiers = qualifiers;
+	}
+
+	@Override
+	public void open() throws IOException {
+		if (isOpen)
+			return; // silently return
+		try {
+			tbl = EagleConfigFactory.load().getHTable(ed.getTable());
+		} catch (RuntimeException ex) {
+			throw new IOException(ex);
+		}
+		final byte[] family = ed.getColumnFamily().getBytes();
+        List<Get> gets = new ArrayList<>(this.rowkeys.size());
+
+        for(byte[] rowkey:rowkeys) {
+            Get get = new Get(rowkey);
+            get.addFamily(family);
+
+            if(qualifiers != null) {
+                for(byte[] qualifier: qualifiers){
+                    get.addColumn(family,qualifier);
+                }
+            }
+
+            gets.add(get);
+        }
+
+        entityResult = tbl.get(gets);
+		isOpen = true;
+	}
+
+	@Override
+	public void close() throws IOException {
+		if(tbl != null){
+			new HTableFactory().releaseHTableInterface(tbl);
+		}
+	}
+
+	@Override
+	public InternalLog read() throws IOException {
+        if(entityResult == null || entityResult.length == 0 || this.getIndex >= entityResult.length - 1){
+            return null;
+        }
+        getIndex ++;
+		InternalLog t = HBaseInternalLogHelper.parse(ed, entityResult[getIndex], this.qualifiers);
+		return t;
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java
new file mode 100755
index 0000000..8ff3448
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexLogReader.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.NavigableMap;
+
+import org.apache.eagle.common.config.EagleConfigFactory;
+import org.apache.eagle.log.entity.HBaseInternalLogHelper;
+import org.apache.eagle.log.entity.InternalLog;
+import org.apache.eagle.log.entity.meta.IndexDefinition;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.HTableFactory;
+import org.apache.hadoop.hbase.client.HTableInterface;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.filter.Filter;
+
+public class UniqueIndexLogReader extends IndexLogReader {
+
+	private final IndexDefinition indexDef;
+	private final List<byte[]> indexRowkeys; 
+	private final byte[][] qualifiers;
+	private final Filter filter;
+	private HTableInterface tbl;
+	private boolean isOpen = false;
+	private Result[] entityResults;
+	private int index = -1;
+
+	public UniqueIndexLogReader(IndexDefinition indexDef, List<byte[]> indexRowkeys, byte[][] qualifiers, Filter filter) {
+		this.indexDef = indexDef;
+		this.indexRowkeys = indexRowkeys;
+		this.qualifiers = qualifiers;
+		this.filter = filter;
+	}
+
+	@Override
+	public void open() throws IOException {
+		if (isOpen)
+			return; // silently return
+		try {
+			tbl = EagleConfigFactory.load().getHTable(indexDef.getEntityDefinition().getTable());
+		} catch (RuntimeException ex) {
+			throw new IOException(ex);
+		}
+		final byte[] family = indexDef.getEntityDefinition().getColumnFamily().getBytes();
+        final List<Get> indexGets = new ArrayList<>();
+        for (byte[] rowkey : indexRowkeys) {
+            Get get = new Get(rowkey);
+            // Return all index qualifiers
+            get.addFamily(family);
+            indexGets.add(get);
+        }
+        final Result[] indexResults = tbl.get(indexGets);
+        indexGets.clear();
+        for (Result indexResult : indexResults) {
+        	final NavigableMap<byte[], byte[]> map = indexResult.getFamilyMap(family);
+        	if (map == null) {
+        		continue;
+        	}
+        	for (byte[] entityRowkey : map.keySet()) {
+                Get get = new Get(entityRowkey);
+                if (filter != null) {
+                	get.setFilter(filter);
+                }
+				if(qualifiers == null) {
+					// filter all qualifiers if output qualifiers are null
+					get.addFamily(family);
+				}else {
+					for (int i = 0; i < qualifiers.length; ++i) {
+						// Return the specified qualifiers
+						get.addColumn(family, qualifiers[i]);
+					}
+				}
+				workaroundHBASE2198(get, filter,qualifiers);
+        		indexGets.add(get);
+        	}
+        }
+        entityResults = tbl.get(indexGets);
+		isOpen = true;
+	}
+
+	@Override
+	public void close() throws IOException {
+		if(tbl != null){
+			new HTableFactory().releaseHTableInterface(tbl);
+		}
+	}
+
+	@Override
+	public InternalLog read() throws IOException {
+		if (entityResults == null) {
+			throw new IllegalArgumentException("entityResults haven't been initialized before reading");
+		}
+		InternalLog t = null;
+		while (entityResults.length > ++index) {
+			Result r = entityResults[index];
+			if (r != null) {
+				if (r.getRow() == null) {
+					continue;
+				}
+				t = HBaseInternalLogHelper.parse(indexDef.getEntityDefinition(), r, qualifiers);
+				break;
+			}
+		}
+		return t;
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java
new file mode 100755
index 0000000..0391d57
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/index/UniqueIndexStreamReader.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.index;
+
+import org.apache.eagle.log.entity.HBaseInternalLogHelper;
+import org.apache.eagle.log.entity.LogReader;
+import org.apache.eagle.log.entity.SearchCondition;
+import org.apache.eagle.log.entity.meta.EntityDefinition;
+import org.apache.eagle.log.entity.meta.IndexDefinition;
+import org.apache.eagle.log.entity.meta.IndexDefinition.IndexType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class UniqueIndexStreamReader extends IndexStreamReader {
+	public UniqueIndexStreamReader(IndexDefinition indexDef, SearchCondition condition) {
+		super(indexDef, condition, new ArrayList<byte[]>());
+		final IndexType type = indexDef.canGoThroughIndex(condition.getQueryExpression(), indexRowkeys);
+		if (!IndexType.UNIQUE_INDEX.equals(type)) {
+			throw new IllegalArgumentException("This query can't go through index: " + condition.getQueryExpression());
+		}
+	}
+
+	public UniqueIndexStreamReader(IndexDefinition indexDef, SearchCondition condition, List<byte[]> indexRowkeys) {
+		super(indexDef, condition, indexRowkeys);
+	}
+
+	@Override
+	protected LogReader createIndexReader() {
+		final EntityDefinition entityDef = indexDef.getEntityDefinition();
+//		final
+		byte[][] outputQualifiers = null;
+		if(!condition.isOutputAll()) {
+			outputQualifiers = HBaseInternalLogHelper.getOutputQualifiers(entityDef, condition.getOutputFields());
+		}
+		return new UniqueIndexLogReader(indexDef, indexRowkeys, outputQualifiers, condition.getFilter());
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java
new file mode 100755
index 0000000..cf40e31
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/BooleanSerDeser.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+/**
+ * @since : 7/3/14,2014
+ */
+public class BooleanSerDeser implements EntitySerDeser<Boolean> {
+
+	public BooleanSerDeser(){}
+
+	@Override
+	public Boolean deserialize(byte[] bytes){
+		if(bytes != null && bytes.length > 0){
+			if(bytes[0] == 0){
+				return false;
+			}else if(bytes[0] == 1){
+				return true;
+			}
+		}
+		return null;
+	}
+
+	@Override
+	public byte[] serialize(Boolean obj){
+		if(obj != null){
+			if(obj){
+				return new byte[]{1};
+			}else{
+				return new byte[]{0};
+			}
+		}
+		return null;
+	}
+
+	@Override
+	public Class<Boolean> type() {
+		return Boolean.class;
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java
new file mode 100644
index 0000000..b64e528
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Column.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ElementType.FIELD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Column {
+	String value() default "";
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java
new file mode 100644
index 0000000..6e3e9c6
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/ColumnFamily.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+@Target({ElementType.TYPE})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface ColumnFamily {
+	String value() default "f";
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java
new file mode 100644
index 0000000..27b011c
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/Double2DArraySerDeser.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+import org.apache.eagle.common.ByteUtil;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+/**
+ * @since 7/22/15
+ */
+public class Double2DArraySerDeser implements EntitySerDeser<double[][]> {
+    private final int SIZE = 8;
+    @Override
+    public double[][] deserialize(byte[] bytes){
+//        if((bytes.length-4) % SIZE != 0)
+//            return null;
+        int offset = 0;
+        // get size of int array
+        int rowSize = ByteUtil.bytesToInt(bytes, offset);
+        offset += 4;
+
+        double[][] data = new double[rowSize][];
+        for(int i=0; i<rowSize; i++) {
+            int colSize = ByteUtil.bytesToInt(bytes, offset);
+            offset += 4;
+            double[] values = null;
+            if (colSize >= 0){
+                values = new double[colSize];
+                for (int j = 0; j < colSize; j++) {
+                    values[j] = ByteUtil.bytesToDouble(bytes, offset);
+                    offset += SIZE;
+                }
+            }
+            data[i] = values;
+        }
+
+        return data;
+    }
+
+    /**
+     *
+     * @param obj
+     * @return
+     */
+    @Override
+    public byte[] serialize(double[][] obj){
+        if(obj == null) return null;
+        ByteArrayOutputStream data = new ByteArrayOutputStream();
+        int size = obj.length;
+        byte[] sizeBytes = ByteUtil.intToBytes(size);
+        data.write(sizeBytes,0,sizeBytes.length);
+
+        try{
+            for(double[] o:obj){
+                if(o!=null){
+                    data.write(ByteUtil.intToBytes(o.length));
+                    for(double d:o){
+                        data.write(ByteUtil.doubleToBytes(d),0,SIZE);
+                    }
+                }else{
+                    data.write(ByteUtil.intToBytes(-1),0,4);
+                }
+            }
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+
+        byte[] bytes = data.toByteArray();
+        try {
+            data.close();
+        } catch (IOException e) {
+            throw new IllegalStateException(e);
+        }
+        return bytes;
+    }
+
+    @Override
+    public Class<double[][]> type() {
+        return double[][].class;
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java
new file mode 100755
index 0000000..d87e31c
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleArraySerDeser.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+import org.apache.eagle.common.ByteUtil;
+
+public class DoubleArraySerDeser implements EntitySerDeser<double[]>{
+
+	public DoubleArraySerDeser(){}
+
+	private final int SIZE = 8;
+		@Override
+		public double[] deserialize(byte[] bytes){
+			if((bytes.length-4) % SIZE != 0)
+				return null;
+			int offset = 0;
+			// get size of int array
+			int size = ByteUtil.bytesToInt(bytes, offset);
+			offset += 4;
+			double[] values = new double[size];
+			for(int i=0; i<size; i++){
+				values[i] = ByteUtil.bytesToDouble(bytes, offset);
+				offset += SIZE;
+			}
+			return values;
+		}
+		
+		/**
+		 * 
+		 * @param obj
+		 * @return
+		 */
+		@Override
+		public byte[] serialize(double[] obj){
+			if(obj == null)
+				return null;
+			int size = obj.length;
+			byte[] array = new byte[4 + SIZE*size];
+			byte[] first = ByteUtil.intToBytes(size);
+			int offset = 0;
+			System.arraycopy(first, 0, array, offset, first.length);
+			offset += first.length;
+			for(int i=0; i<size; i++){
+				System.arraycopy(ByteUtil.doubleToBytes(obj[i]), 0, array, offset, SIZE);
+				offset += SIZE;
+			}
+			return array;
+		}
+
+	@Override
+	public Class<double[]> type() {
+		return double[].class;
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java
new file mode 100755
index 0000000..330a99d
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/DoubleSerDeser.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+import org.apache.eagle.common.ByteUtil;
+
+public class DoubleSerDeser implements EntitySerDeser<Double>{
+
+	@Override
+	public Double deserialize(byte[] bytes){
+		if(bytes.length < 8)
+			return null;
+		return ByteUtil.bytesToDouble(bytes);
+	}
+	
+	@Override
+	public byte[] serialize(Double obj){
+		if(obj == null)
+			return null;
+		return ByteUtil.doubleToBytes(obj);
+	}
+
+	@Override
+	public Class<Double> type(){
+		return Double.class;
+	}
+}

http://git-wip-us.apache.org/repos/asf/incubator-eagle/blob/afe86834/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityConstants.java
----------------------------------------------------------------------
diff --git a/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityConstants.java b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityConstants.java
new file mode 100644
index 0000000..930743e
--- /dev/null
+++ b/eagle-core/eagle-query/eagle-entity-base/src/main/java/org/apache/eagle/log/entity/meta/EntityConstants.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.eagle.log.entity.meta;
+
+import org.apache.eagle.common.DateTimeUtil;
+
+public class EntityConstants {
+	
+	public static final String FIXED_WRITE_HUMANTIME = "1970-01-02 00:00:00";
+	public static final String FIXED_READ_START_HUMANTIME = "1970-01-01 00:00:00";
+	public static final String FIXED_READ_END_HUMANTIME = "1970-01-03 00:00:00";
+	
+	public static final long FIXED_WRITE_TIMESTAMP = 
+			DateTimeUtil.humanDateToSecondsWithoutException(FIXED_WRITE_HUMANTIME) * 1000;
+
+}