public class DeepEquals implements Function<Cells, Boolean> {

	private static final long serialVersionUID = -6143471452730703044L;

	/**
	 * Term to be compared.
	 */
	private final Term<?> term;


	/**
	 * Column cell to compare to.
	 */
	private final ColumnName column;

	/**
	 * Basic constructor for the DeepEquals function class.
	 * @param column
	 * 				Column cell to compare to
	 * @param term
	 * 				Term to be compared
	 */
	public DeepEquals(ColumnName column, Term<?> term) {
		this.term = term;
		this.column = column;
	}

	@Override
	public Boolean call(Cells cells) {

		Object currentValue = cells.getCellByName(column.getTableName().getQualifiedName(), column.getName())
				.getCellValue();

		return term.equals(currentValue);
	}

} public class GreaterEqualThan implements Function<Cells, Boolean> {

	private static final long serialVersionUID = 2927596112428729111L;

	/**
	 * Value to be compared.
	 */
	private final Term<?> term;

	/**
	 * Column cell to compare to.
	 */
	private final ColumnName column;

	/**
	 * Basic constructor for the GreaterEqualThan function class.
	 * 
	 * @param column
	 * 				Column cell to compare to
	 * @param term
	 * 				Value to be compared
	 */
	public GreaterEqualThan(ColumnName column, Term term) {
		this.term = term;
		this.column = column;
	}

	@Override
	public Boolean call(Cells cells) {
		Object obj = cells.getCellByName(column.getTableName().getQualifiedName(), column.getName()).getCellValue();
		return ((Comparable) term).compareTo(obj) <= 0;
	}
} public class GreaterThan implements Function<Cells, Boolean> {

	private static final long serialVersionUID = 2675616112608139116L;

	/**
	 * Term to be compared.
	 */
	private final Term<?> term;

	/**
	 * Column cell to compare to.
	 */
	private final ColumnName column;

	/**
	 * Basic constructor for the GreaterThan function class.
	 * 
	 * @param column
	 * 				Column cell to compare to
	 * @param term
	 * 				Term to be compared
	 */
	public GreaterThan(ColumnName column, Term term) {
		this.term = term;
		this.column = column;
	}

	@Override
	public Boolean call(Cells cells) {
		Object obj = cells.getCellByName(column.getTableName().getQualifiedName(), column.getName()).getCellValue();
		return ((Comparable) term).compareTo(obj) < 0;
	}
} public class LessEqualThan implements Function<Cells, Boolean> {

	private static final long serialVersionUID = 2775666112428131116L;

	/**
	 * Term to be compared.
	 */
	private final Term<?> term;

	/**
	 * Column cell to compare to.
	 */
	private final ColumnName column;

	/**
	 * Basic constructor for the LessEqualThan function class.
	 * 
	 * @param column
	 * 				Column cell to compare to
	 * @param term
	 * 				Term to be compared
	 */
	public LessEqualThan(ColumnName column, Term term) {
		this.term = term;
		this.column = column;
	}

	@Override
	public Boolean call(Cells cells) {
		Object obj = cells.getCellByName(column.getTableName().getQualifiedName(), column.getName()).getCellValue();
		return ((Comparable) term).compareTo(obj) >= 0;

	}
} public class LessThan implements Function<Cells, Boolean> {

	private static final long serialVersionUID = 2675616112608139116L;

	/**
	 * Term to be compared.
	 */
	private final Term<?> term;

	/**
	 * Column cell to compare to.
	 */
	private final ColumnName column;

	/**
	 * Basic constructor for the LessThan function class.
	 * 
	 * @param column
	 * 				Column cell to compare to
	 * @param term
	 * 				Term to be compared
	 */
	public LessThan(ColumnName column, Term term) {
		this.term = term;
		this.column = column;
	}

	@Override
	public Boolean call(Cells cells) {
		Object obj = cells.getCellByName(column.getTableName().getQualifiedName(), column.getName()).getCellValue();
		return ((Comparable) term).compareTo(obj) > 0;
	}
} public class NotEquals implements Function<Cells, Boolean> {

	private static final long serialVersionUID = 927384912608139416L;

	/**
	 * Term to be compared.
	 */
	private final Term<?> term;

	/**
	 * Column cell to compare to.
	 */
	private final ColumnName column;

	/**
	 * Basic constructor for the NotEquals function class.
	 * 
	 * @param column
	 * 				Column cell to compare to
	 * @param term
	 * 				Term to be compared
	 */
	public NotEquals(ColumnName column, Term term) {
		this.term = term;
		this.column = column;
	}

	@Override
	public Boolean call(Cells cells) {
		Object currentValue = cells.getCellByName(column.getTableName().getQualifiedName(), column.getName())
				.getCellValue();
		return !term.equals(currentValue);
	}
} public class OrderByComparator implements Comparator<Cells>,Serializable {

	private static final long serialVersionUID = 432384912608139416L;

	/**
	 * Order clauses for the OrderBy function.
	 */
	private List<OrderByClause> orderByClauses;

	/**
	 * Basic constructor for the OrderBy function class.
	 * @param orderByClauses
	 * 						Order clauses for the OrderBy function
	 */
	public OrderByComparator(List<OrderByClause> orderByClauses) {
		this.orderByClauses = orderByClauses;
	}


	@Override
	public int compare(Cells o1, Cells o2) {

		int result = 0;

		for (OrderByClause orderByClause : orderByClauses) {

			ColumnSelector columnSelector = (ColumnSelector) orderByClause.getSelector();

			Cell cell1 = o1.getCellByName(columnSelector.getName().getTableName().getQualifiedName(),
					columnSelector.getName().getName());

			Cell cell2 = o2.getCellByName(columnSelector.getName().getTableName().getQualifiedName(),
					columnSelector.getName().getName());

			OrderDirection order = orderByClause.getDirection();

			if(order == OrderDirection.ASC){
				result = ((Comparable) cell1.getCellValue()).compareTo(cell2.getValue());
			}else{
				result = ((Comparable) cell2.getCellValue()).compareTo(cell1.getValue());
			}
			if(result!=0){
				break;
			}

		}

		return result;

	}
} public class BooleanTerm extends Term<Boolean> {
	private static final long serialVersionUID = 2872212148572680680L;

	/**
	 * Constructor from a String representation of a Boolean value.
	 * 
	 * @param term
	 *            The string representation of a Boolean value
	 */
	public BooleanTerm(String term) {
		super(Boolean.class, Boolean.valueOf(term));
	}
} public class DoubleTerm extends Term<Double> {

	private static final long serialVersionUID = -578510540271635667L;

	/**
	 * Constructor from a String representation of a Double value.
	 * @param term
	 * 				String representation of a Double value
	 */
	public DoubleTerm(String term) {
		super(Double.class, Double.valueOf(term));
	}

	/**
	 * Constructor from Double type.
	 * @param term
	 * 				Double value
	 */
	public DoubleTerm(Double term) {
		super(Double.class, term);
	}
} public class LongTerm extends Term<Long> {

	private static final long serialVersionUID = 7097178218828822792L;

	/**
	 * Constructor from a String representation of a Long value.
	 * @param term
	 * 				String representation of a Long value
	 */
	public LongTerm(String term) {
		super(Long.class, Long.valueOf(term));
	}

	/**
	 * Constructor from Long type.
	 * @param term
	 * 				The Long value
	 */
	public LongTerm(Long term) {
		super(Long.class, term);
	}
} public class StringTerm extends Term<String> {
	private static final long serialVersionUID = 4470491967411363431L;
	private boolean quotedLiteral = false;

	/**
	 * Basic constructor.
	 * @param term
	 * 				String value
	 * @param quotedLiteral
	 * 				Defines whether the String is quoted or not
	 * 				
	 */
	public StringTerm(String term, boolean quotedLiteral) {
		super(String.class, term);
		this.type = TYPE_TERM;
		this.quotedLiteral = quotedLiteral;
	}

	/**
	 * Constructor from String type.
	 * @param term
	 * 				String value
	 *	
	 */
	public StringTerm(String term) {
		this(term, false);
	}

	/**
	 * Constructor from Term<String>.
	 * @param term
	 * 				Term of the type String 
	 * 						
	 */
	public StringTerm(Term<String> term) {
		super(String.class, term.getTermValue().toString());
	}

	public boolean isQuotedLiteral() {
		return quotedLiteral;
	}

	/** {@inheritDoc} */
	@Override
	public String toString() {
		if (this.isQuotedLiteral()) {
			return "'" + value + "'";
		} else {
			return value;
		}
	}
} public abstract class Term<T extends Comparable<T>> extends ValueCell<T> implements Comparable<T>,
Serializable {

	/**
	 * Class logger.
	 */
	private static final Logger LOG = Logger.getLogger(Term.class);

	private static final long serialVersionUID = -4258938152892510227L;
	protected Class<? extends Comparable<?>> clazz;
	protected T value;

	/**
	 * Basic constructor.
	 * @param clazz
	 * 				Class of the type that extends Comparable
	 * @param value
	 * 				Value of the Term
	 */
	public Term(Class<? extends Comparable<?>> clazz, T value) {
		this.clazz = clazz;
		this.value = value;
	}

	/**
	 * Get the Term Java Class.
	 * 
	 * @return A {@link java.lang.Class}.
	 */
	public Class<? extends Comparable<?>> getTermClass() {
		return clazz;
	}

	/**
	 * Get the term value.
	 * 
	 * @return A {@link java.lang.Object} with the value.
	 */
	public T getTermValue() {
		return value;
	}

	@Override
	public String getStringValue() {
		return value.toString();
	}

	@Override
	public String toString() {
		return value.toString();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.lang.Comparable#compareTo(java.lang.Object)
	 */
	@Override
	public int compareTo(T o) {

		if (!(this.clazz.isInstance(o))) {

			try {
				if ((this.clazz).equals(Long.class) && o.getClass().equals(Integer.class)) {
					Long obj = Long.valueOf(o.toString());
					return Long.valueOf(this.getStringValue()).compareTo(obj);
				} else if ((this.clazz).equals(Double.class) && o.getClass().equals(Float.class)) {
					Double obj = Double.valueOf(o.toString());
					return Double.valueOf(this.getStringValue()).compareTo(obj);
				}
			} catch (NumberFormatException | ClassCastException e) {
				LOG.error("Sorry, unable to Cast incompatible types ->" + this.clazz + " & " + o.getClass(), e);
			}
			return this.value.compareTo(o);
		}
		return this.value.compareTo(o);
	}

	/**
	 * Returns a hash code value for the object. This method is supported for the benefit of hash tables such as those
	 * provided by {@link java.util.HashMap}.
	 * 
	 * @return a hash code value for this object.
	 * @see Object#equals(Object)
	 * @see System#identityHashCode
	 */
	@Override
	public int hashCode() {
		return clazz.hashCode() * this.getTermValue().hashCode();
	}

	/**
	 * Indicates whether some other object is "equal to" this one.
	 * <p/>
	 * The {@code equals} method implements an equivalence relation on non-null object references:
	 * 
	 * @param obj
	 *            the reference object with which to compare.
	 * @return {@code true} if this object is the same as the obj argument; {@code false} otherwise.
	 * @see #hashCode()
	 * @see java.util.HashMap
	 */
	@Override
	public boolean equals(Object obj) {
		if (obj == null) {
			return false;
		}
		if (!(this.clazz.isInstance(obj))) {
			try {
				if (obj instanceof String) {
					String stringObj = (String) obj;
					return this.getStringValue().equals(stringObj);
				} else if (obj instanceof Float) {
					Float floatObj = (Float) obj;
					return Float.valueOf(this.getStringValue()).equals(floatObj);
				} else if (obj instanceof Integer) {
					Integer integerObj = (Integer) obj;
					return Integer.valueOf(this.getStringValue()).equals(integerObj);
				} else if (obj instanceof Long) {
					Long longObj = (Long) obj;
					return Long.valueOf(this.getStringValue()).equals(longObj);
				}
			} catch (NumberFormatException e) {
				LOG.error("Sorry, unable to Cast incompatible types ->" + this.clazz + " & " + obj.getClass(), e);
			}
			return super.equals(obj);
		}
		return this.value.equals(obj);
	}
} public class FilterColumns implements Function<Cells, Cells> {

	private static final long serialVersionUID = -6143471789450703044L;

	/**
	 * Columns to be kept.
	 */
	private final List<Selector> columns;

	/**
	 * Basic constructor.
	 * 
	 * @param columns
	 * 				The columns
	 */
	public FilterColumns(List<Selector> columns) {
		this.columns = columns;
	}

	@Override
	public Cells call(Cells cells) {

		Cells cellsOut = new Cells();
		for (Selector columnName : columns) {
			Cell cell = cells.getCellByName(columnName.getColumnName().getTableName().getQualifiedName(),
					columnName.getColumnName().getName());
			cellsOut.add(columnName.getColumnName().getTableName().getQualifiedName(), cell);
		}

		return cellsOut;
	}
} public class JoinCells implements Function<Tuple2<List<Object>, Tuple2<Cells, Cells>>, Cells> {

	private static final long serialVersionUID = 4534397129761833793L;

	public JoinCells() {

	}

	@Override
	public Cells call(Tuple2<List<Object>, Tuple2<Cells, Cells>> result) {

		Cells left = result._2()._1();
		Cells right = result._2()._2();

		Cells joinedCells = new Cells();
		joinedCells.addAll(left.getInternalCells());
		joinedCells.addAll(right.getInternalCells());

		return joinedCells;
	}
} public class MapKeyForJoin implements PairFunction<Cells, List<Object>, Cells> {

	private static final long serialVersionUID = -6677647619149716567L;

	/**
	 * Map key.
	 */
	private final List<ColumnName> keys;

	/**
	 * MapKeyForJoin maps a field in a Cell.
	 * 
	 * @param keys
	 *            Field to map
	 */
	public MapKeyForJoin(List<ColumnName> keys) {
		this.keys = keys;
	}

	@Override
	public Tuple2<List<Object>, Cells> call(Cells cells) {

		List<Object> cellsValues = new LinkedList<>();
		for (ColumnName columnKey : keys) {
			String tableName = columnKey.getTableName().getQualifiedName();
			Cell cell = cells.getCellByName(tableName, columnKey.getName());
			cellsValues.add(cell.getCellValue());
		}

		return new Tuple2<>(cellsValues, cells);
	}
}
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import com.stratio.connector.deep.engine.query.functions.DeepEquals;
import com.stratio.connector.deep.engine.query.functions.GreaterEqualThan;
import com.stratio.connector.deep.engine.query.functions.GreaterThan;
import com.stratio.connector.deep.engine.query.functions.LessEqualThan;
import com.stratio.connector.deep.engine.query.functions.LessThan;
import com.stratio.connector.deep.engine.query.functions.NotEquals;
import com.stratio.connector.deep.engine.query.functions.OrderByComparator;
import com.stratio.connector.deep.engine.query.structures.BooleanTerm;
import com.stratio.connector.deep.engine.query.structures.DoubleTerm;
import com.stratio.connector.deep.engine.query.structures.LongTerm;
import com.stratio.connector.deep.engine.query.structures.StringTerm;
import com.stratio.connector.deep.engine.query.structures.Term;
import com.stratio.connector.deep.engine.query.transformation.FilterColumns;
import com.stratio.connector.deep.engine.query.transformation.JoinCells;
import com.stratio.connector.deep.engine.query.transformation.MapKeyForJoin;
import com.stratio.crossdata.common.data.ColumnName;
import com.stratio.crossdata.common.exceptions.ExecutionException;
import com.stratio.crossdata.common.exceptions.UnsupportedException;
import com.stratio.crossdata.common.statements.structures.BooleanSelector;
import com.stratio.crossdata.common.statements.structures.ColumnSelector;
import com.stratio.crossdata.common.statements.structures.FloatingPointSelector;
import com.stratio.crossdata.common.statements.structures.IntegerSelector;
import com.stratio.crossdata.common.statements.structures.Operator;
import com.stratio.crossdata.common.statements.structures.OrderByClause;
import com.stratio.crossdata.common.statements.structures.Relation;
import com.stratio.crossdata.common.statements.structures.Selector;
import com.stratio.crossdata.common.statements.structures.SelectorType;
import com.stratio.crossdata.common.statements.structures.StringSelector;
import com.stratio.deep.commons.entity.Cell;
import com.stratio.deep.commons.entity.Cells;
import com.stratio.deep.commons.filter.FilterType;
import scala.Tuple2;
/*
 * Licensed to STRATIO (C) under one or more contributor license agreements.
 * See the NOTICE file distributed with this work for additional information
 * regarding copyright ownership.  The STRATIO (C) 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 com.stratio.connector.deep.engine.query;





/**
 * Utils for the query Filters.
 */
public final class QueryFilterUtils {

	/**
	 * Class logger.
	 */
	private static transient final Logger LOGGER = Logger.getLogger(QueryFilterUtils.class);


	private QueryFilterUtils() {
	}

	/**
	 * Take a RDD and a Relation and apply suitable filter to the RDD. Execute where clause on Deep.
	 * 
	 * @param rdd
	 *            RDD which filter must be applied.
	 * @param relation
	 *            {@link com.stratio.crossdata.common.statements.structures.relationships.Relation} to apply.
	 * @return A new RDD with the result.
	 * @throws UnsupportedException
	 */
	static JavaRDD<Cells> doWhere(JavaRDD<Cells> rdd, Relation relation) throws UnsupportedException,
	ExecutionException {

        if (!relation.getOperator().isInGroup(Operator.Group.COMPARATOR)) {
            throw new ExecutionException("Unknown Filter found [" + relation.getOperator().toString()+ "]");
        }

		Operator operator = relation.getOperator();
		JavaRDD<Cells> result;
		ColumnName column = ((ColumnSelector) relation.getLeftTerm()).getName();
		Term rightTerm = filterFromRightTermWhereRelation(relation);
        Function function = chooseCompareFunction(operator, column, rightTerm);
        result = rdd.filter(function);
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("Rdd["+result.id()+"]: Rdd["+rdd.id()+"].filter("+function.getClass().getSimpleName()+")");
        }


		return result;
	}

    private static Function chooseCompareFunction(Operator operator, ColumnName column, Term rightTerm)
            throws UnsupportedException {
        Function function = null;

        switch (operator) {
        case EQ:
            function = new DeepEquals(column, rightTerm);
            break;
        case DISTINCT:
            function = new NotEquals(column, rightTerm);
            break;
        case GT:
            function = new GreaterThan(column, rightTerm);
            break;
        case GET:
            function = new GreaterEqualThan(column, rightTerm);
            break;
        case LT:
            function = new LessThan(column, rightTerm);
            break;
        case LET:
            function = new LessEqualThan(column, rightTerm);
            break;

        default:

            String message = "Operator not supported: " + operator;
            LOGGER.error(message);
            throw new UnsupportedException(message);

        }
        return function;
    }

    /**
	 * Build JavaRDD<Cells> from list of Cells and select Columns.
	 * 
	 * @param rdd
	 *            Cells RDD.
	 * @param selectedCols
	 *            Set of fields selected in the SelectStatement.
	 * @return JavaRDD<Cells>
	 */

	static JavaRDD<Cells> filterSelectedColumns(JavaRDD<Cells> rdd, final Set<Selector> selectedCols) {

		List<Selector> list = new ArrayList<>(selectedCols);

		return rdd.map(new FilterColumns(list));
	}

	static JavaRDD<Cells> doJoin(JavaRDD<Cells> leftRdd, JavaRDD<Cells> rightRdd, List<Relation> joinRelations) {

		JavaRDD<Cells> joinedResult = null;

		List<ColumnName> firstTables = new ArrayList<>();
		List<ColumnName> secondTables = new ArrayList<>();


		for (Relation relation : joinRelations) {

			ColumnSelector selectorRight = (ColumnSelector) relation.getRightTerm();
			ColumnSelector selectorLeft = (ColumnSelector) relation.getLeftTerm();

			if (relation.getOperator().equals(Operator.EQ)) {
				firstTables.add(selectorLeft.getName());
				secondTables.add(selectorRight.getName());
				LOGGER.info("INNER JOIN on: " + selectorRight.getName().getName() + " - "
							+ selectorLeft.getName().getName());

			}

		}

		JavaPairRDD<List<Object>, Cells> rddLeft = leftRdd.mapToPair(new MapKeyForJoin(firstTables));
        if (LOGGER.isDebugEnabled()){
            LOGGER.debug("RDD["+rddLeft.id()+"] = RDD["+leftRdd.id()+"].mapToPai()");
        }
		JavaPairRDD<List<Object>, Cells> rddRight = rightRdd.mapToPair(new MapKeyForJoin(secondTables));
        if (LOGGER.isDebugEnabled()){
            LOGGER.debug("RDD["+rddRight.id()+"] = RDD["+rightRdd.id()+"].mapToPai()");
        }
		if (rddLeft != null && rddRight != null) {
			JavaPairRDD<List<Object>, Tuple2<Cells, Cells>> joinRDD = rddLeft.join(rddRight);
            if (LOGGER.isDebugEnabled()){
                LOGGER.debug("RDD["+joinRDD.id()+"] = RDD["+rddLeft.id()+"].join(RDD["+rddRight.id()+"])");
            }
			joinedResult = joinRDD.map(new JoinCells());
            if (LOGGER.isDebugEnabled()){
                LOGGER.debug("RDD["+joinedResult.id()+"] = RDD["+joinRDD.id()+"].map()");
            }
		}

		return joinedResult;

	}

	/**
	 * Method that returns the right term in a relation.
	 * 
	 * @param relation
	 * 					The relation
	 * @return
	 * 			The right term
	 * @throws ExecutionException
	 */
	public static Term filterFromRightTermWhereRelation(Relation relation) throws ExecutionException {

		SelectorType type = relation.getRightTerm().getType();
		Term rightField = null;

		switch (type) {
		case STRING:
			rightField = new StringTerm(((StringSelector) relation.getRightTerm()).getValue());