You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by el...@apache.org on 2016/04/20 00:42:06 UTC
[02/10] phoenix git commit: PHOENIX-2743 HivePhoenixHandler for
big-big join with predicate push down
http://git-wip-us.apache.org/repos/asf/phoenix/blob/537b90be/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexPredicateAnalyzer.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexPredicateAnalyzer.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexPredicateAnalyzer.java
new file mode 100644
index 0000000..659983a
--- /dev/null
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexPredicateAnalyzer.java
@@ -0,0 +1,523 @@
+/*
+ * 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.phoenix.hive.ql.index;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
+import org.apache.hadoop.hive.ql.lib.DefaultGraphWalker;
+import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
+import org.apache.hadoop.hive.ql.lib.Dispatcher;
+import org.apache.hadoop.hive.ql.lib.GraphWalker;
+import org.apache.hadoop.hive.ql.lib.Node;
+import org.apache.hadoop.hive.ql.lib.NodeProcessor;
+import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.lib.Rule;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDescUtils;
+import org.apache.hadoop.hive.ql.plan.ExprNodeFieldDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBaseCompare;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBetween;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFIn;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNot;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotNull;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNull;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToBinary;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToChar;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToDate;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToDecimal;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToUnixTimeStamp;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToUtcTimestamp;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFToVarchar;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Stack;
+
+/**
+ * Clone of org.apache.hadoop.hive.ql.index.IndexPredicateAnalyzer with modifying
+ * analyzePredicate method.
+ *
+ *
+ */
+public class IndexPredicateAnalyzer {
+
+ private static final Log LOG = LogFactory.getLog(IndexPredicateAnalyzer.class);
+
+ private final Set<String> udfNames;
+ private final Map<String, Set<String>> columnToUDFs;
+ private FieldValidator fieldValidator;
+
+ private boolean acceptsFields;
+
+ public IndexPredicateAnalyzer() {
+ udfNames = new HashSet<String>();
+ columnToUDFs = new HashMap<String, Set<String>>();
+ }
+
+ public void setFieldValidator(FieldValidator fieldValidator) {
+ this.fieldValidator = fieldValidator;
+ }
+
+ /**
+ * Registers a comparison operator as one which can be satisfied by an index
+ * search. Unless this is called, analyzePredicate will never find any
+ * indexable conditions.
+ *
+ * @param udfName name of comparison operator as returned by either
+ * {@link GenericUDFBridge#getUdfName} (for simple UDF's) or
+ * udf.getClass().getName() (for generic UDF's).
+ */
+ public void addComparisonOp(String udfName) {
+ udfNames.add(udfName);
+ }
+
+ /**
+ * Clears the set of column names allowed in comparisons. (Initially, all
+ * column names are allowed.)
+ */
+ public void clearAllowedColumnNames() {
+ columnToUDFs.clear();
+ }
+
+ /**
+ * Adds a column name to the set of column names allowed.
+ *
+ * @param columnName name of column to be allowed
+ */
+ public void allowColumnName(String columnName) {
+ columnToUDFs.put(columnName, udfNames);
+ }
+
+ /**
+ * add allowed functions per column
+ *
+ * @param columnName
+ * @param udfs
+ */
+ public void addComparisonOp(String columnName, String... udfs) {
+ Set<String> allowed = columnToUDFs.get(columnName);
+ if (allowed == null || allowed == udfNames) {
+ // override
+ columnToUDFs.put(columnName, new HashSet<String>(Arrays.asList(udfs)));
+ } else {
+ allowed.addAll(Arrays.asList(udfs));
+ }
+ }
+
+ /**
+ * Analyzes a predicate.
+ *
+ * @param predicate predicate to be analyzed
+ * @param searchConditions receives conditions produced by analysis
+ * @return residual predicate which could not be translated to
+ * searchConditions
+ */
+ public ExprNodeDesc analyzePredicate(ExprNodeDesc predicate, final List<IndexSearchCondition>
+ searchConditions) {
+
+ Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
+ NodeProcessor nodeProcessor = new NodeProcessor() {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx, Object...
+ nodeOutputs) throws SemanticException {
+
+ // We can only push down stuff which appears as part of
+ // a pure conjunction: reject OR, CASE, etc.
+ for (Node ancestor : stack) {
+ if (nd == ancestor) {
+ break;
+ }
+ if (!FunctionRegistry.isOpAnd((ExprNodeDesc) ancestor)) {
+ return nd;
+ }
+ }
+
+ return analyzeExpr((ExprNodeGenericFuncDesc) nd, searchConditions, nodeOutputs);
+ }
+ };
+
+ Dispatcher disp = new DefaultRuleDispatcher(nodeProcessor, opRules, null);
+ GraphWalker ogw = new DefaultGraphWalker(disp);
+ ArrayList<Node> topNodes = new ArrayList<Node>();
+ topNodes.add(predicate);
+ HashMap<Node, Object> nodeOutput = new HashMap<Node, Object>();
+
+ try {
+ ogw.startWalking(topNodes, nodeOutput);
+ } catch (SemanticException ex) {
+ throw new RuntimeException(ex);
+ }
+
+ ExprNodeDesc residualPredicate = (ExprNodeDesc) nodeOutput.get(predicate);
+ return residualPredicate;
+ }
+
+ // Check if ExprNodeColumnDesc is wrapped in expr.
+ // If so, peel off. Otherwise return itself.
+ private ExprNodeDesc getColumnExpr(ExprNodeDesc expr) {
+ if (expr instanceof ExprNodeColumnDesc) {
+ return expr;
+ }
+ ExprNodeGenericFuncDesc funcDesc = null;
+ if (expr instanceof ExprNodeGenericFuncDesc) {
+ funcDesc = (ExprNodeGenericFuncDesc) expr;
+ }
+ if (null == funcDesc) {
+ return expr;
+ }
+ GenericUDF udf = funcDesc.getGenericUDF();
+ // check if its a simple cast expression.
+ if ((udf instanceof GenericUDFBridge || udf instanceof GenericUDFToBinary || udf
+ instanceof GenericUDFToChar
+ || udf instanceof GenericUDFToVarchar || udf instanceof GenericUDFToDecimal
+ || udf instanceof GenericUDFToDate || udf instanceof GenericUDFToUnixTimeStamp
+ || udf instanceof GenericUDFToUtcTimestamp) && funcDesc.getChildren().size() == 1
+ && funcDesc.getChildren().get(0) instanceof ExprNodeColumnDesc) {
+ return expr.getChildren().get(0);
+ }
+ return expr;
+ }
+
+ private void processingBetweenOperator(ExprNodeGenericFuncDesc expr,
+ List<IndexSearchCondition> searchConditions, Object...
+ nodeOutputs) {
+ ExprNodeColumnDesc columnDesc = null;
+ String[] fields = null;
+
+ if (nodeOutputs[1] instanceof ExprNodeFieldDesc) {
+ // rowKey field
+ ExprNodeFieldDesc fieldDesc = (ExprNodeFieldDesc) nodeOutputs[1];
+ fields = ExprNodeDescUtils.extractFields(fieldDesc);
+
+ ExprNodeDesc[] extracted = ExprNodeDescUtils.extractComparePair((ExprNodeDesc)
+ nodeOutputs[1], (ExprNodeDesc) nodeOutputs[2]);
+ columnDesc = (ExprNodeColumnDesc) extracted[0];
+ } else if (nodeOutputs[0] instanceof ExprNodeGenericFuncDesc) {
+ columnDesc = (ExprNodeColumnDesc) ((ExprNodeGenericFuncDesc) nodeOutputs[1])
+ .getChildren().get(0);
+ } else {
+ columnDesc = (ExprNodeColumnDesc) nodeOutputs[1];
+ }
+
+ String udfName = expr.getGenericUDF().getUdfName();
+ ExprNodeConstantDesc[] betweenConstants = new ExprNodeConstantDesc[]{
+ (ExprNodeConstantDesc) nodeOutputs[2], (ExprNodeConstantDesc) nodeOutputs[3]};
+ boolean isNot = (Boolean) ((ExprNodeConstantDesc) nodeOutputs[0]).getValue();
+
+ searchConditions.add(new IndexSearchCondition(columnDesc, udfName, betweenConstants,
+ expr, fields, isNot));
+ }
+
+ private void processingInOperator(ExprNodeGenericFuncDesc expr, List<IndexSearchCondition>
+ searchConditions, boolean isNot, Object... nodeOutputs) {
+ ExprNodeColumnDesc columnDesc = null;
+ String[] fields = null;
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Processing In Operator. nodeOutputs : " + Lists.newArrayList(nodeOutputs));
+ }
+
+ if (nodeOutputs[0] instanceof ExprNodeFieldDesc) {
+ // rowKey field
+ ExprNodeFieldDesc fieldDesc = (ExprNodeFieldDesc) nodeOutputs[0];
+ fields = ExprNodeDescUtils.extractFields(fieldDesc);
+
+ ExprNodeDesc[] extracted = ExprNodeDescUtils.extractComparePair((ExprNodeDesc)
+ nodeOutputs[0], (ExprNodeDesc) nodeOutputs[1]);
+
+ if (extracted == null) { // adding for tez
+ return;
+ }
+
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("nodeOutputs[0] : " + nodeOutputs[0] + ", nodeOutputs[1] : " +
+ nodeOutputs[1] + " => " + Lists.newArrayList(extracted));
+ }
+
+ columnDesc = (ExprNodeColumnDesc) extracted[0];
+ } else if (nodeOutputs[0] instanceof ExprNodeGenericFuncDesc) {
+ columnDesc = (ExprNodeColumnDesc) ((ExprNodeGenericFuncDesc) nodeOutputs[0])
+ .getChildren().get(0);
+ } else {
+ columnDesc = (ExprNodeColumnDesc) nodeOutputs[0];
+ }
+
+ String udfName = expr.getGenericUDF().getUdfName();
+ ExprNodeConstantDesc[] inConstantDescs = new ExprNodeConstantDesc[nodeOutputs.length - 1];
+
+ for (int i = 0, limit = inConstantDescs.length; i < limit; i++) {
+ if (!(nodeOutputs[i + 1] instanceof ExprNodeConstantDesc)) { // adding for tez
+ return;
+ }
+
+ inConstantDescs[i] = (ExprNodeConstantDesc) nodeOutputs[i + 1];
+ }
+
+ searchConditions.add(new IndexSearchCondition(columnDesc, udfName, inConstantDescs, expr,
+ fields, isNot));
+ }
+
+ private void processingNullOperator(ExprNodeGenericFuncDesc expr, List<IndexSearchCondition>
+ searchConditions, Object... nodeOutputs) {
+ ExprNodeColumnDesc columnDesc = null;
+ String[] fields = null;
+
+ if (nodeOutputs[0] instanceof ExprNodeFieldDesc) {
+ // rowKey field
+ ExprNodeFieldDesc fieldDesc = (ExprNodeFieldDesc) nodeOutputs[0];
+ fields = ExprNodeDescUtils.extractFields(fieldDesc);
+
+ ExprNodeDesc[] extracted = ExprNodeDescUtils.extractComparePair((ExprNodeDesc)
+ nodeOutputs[0], new ExprNodeConstantDesc());
+ columnDesc = (ExprNodeColumnDesc) extracted[0];
+ } else if (nodeOutputs[0] instanceof ExprNodeGenericFuncDesc) {
+ columnDesc = (ExprNodeColumnDesc) ((ExprNodeGenericFuncDesc) nodeOutputs[0])
+ .getChildren().get(0);
+ } else {
+ columnDesc = (ExprNodeColumnDesc) nodeOutputs[0];
+ }
+
+ String udfName = expr.getGenericUDF().getUdfName();
+
+ searchConditions.add(new IndexSearchCondition(columnDesc, udfName, null, expr, fields,
+ false));
+ }
+
+ private void processingNotNullOperator(ExprNodeGenericFuncDesc expr,
+ List<IndexSearchCondition> searchConditions, Object...
+ nodeOutputs) {
+ ExprNodeColumnDesc columnDesc = null;
+ String[] fields = null;
+
+ if (nodeOutputs[0] instanceof ExprNodeFieldDesc) {
+ // rowKey field
+ ExprNodeFieldDesc fieldDesc = (ExprNodeFieldDesc) nodeOutputs[0];
+ fields = ExprNodeDescUtils.extractFields(fieldDesc);
+
+ ExprNodeDesc[] extracted = ExprNodeDescUtils.extractComparePair((ExprNodeDesc)
+ nodeOutputs[0], new ExprNodeConstantDesc());
+ columnDesc = (ExprNodeColumnDesc) extracted[0];
+ } else if (nodeOutputs[0] instanceof ExprNodeGenericFuncDesc) {
+ columnDesc = (ExprNodeColumnDesc) ((ExprNodeGenericFuncDesc) nodeOutputs[0])
+ .getChildren().get(0);
+ } else {
+ columnDesc = (ExprNodeColumnDesc) nodeOutputs[0];
+ }
+
+ String udfName = expr.getGenericUDF().getUdfName();
+
+ searchConditions.add(new IndexSearchCondition(columnDesc, udfName, null, expr, fields,
+ true));
+ }
+
+ private ExprNodeDesc analyzeExpr(ExprNodeGenericFuncDesc expr, List<IndexSearchCondition>
+ searchConditions, Object... nodeOutputs) throws SemanticException {
+
+ if (FunctionRegistry.isOpAnd(expr)) {
+ assert (nodeOutputs.length == 2);
+ ExprNodeDesc residual1 = (ExprNodeDesc) nodeOutputs[0];
+ ExprNodeDesc residual2 = (ExprNodeDesc) nodeOutputs[1];
+ if (residual1 == null) {
+ return residual2;
+ }
+ if (residual2 == null) {
+ return residual1;
+ }
+ List<ExprNodeDesc> residuals = new ArrayList<ExprNodeDesc>();
+ residuals.add(residual1);
+ residuals.add(residual2);
+ return new ExprNodeGenericFuncDesc(TypeInfoFactory.booleanTypeInfo, FunctionRegistry
+ .getGenericUDFForAnd(), residuals);
+ }
+
+ GenericUDF genericUDF = expr.getGenericUDF();
+ if (!(genericUDF instanceof GenericUDFBaseCompare)) {
+ // 2015-10-22 Added by JeongMin Ju : Processing Between/In Operator
+ if (genericUDF instanceof GenericUDFBetween) {
+ // In case of not between, The value of first element of nodeOutputs is true.
+ // otherwise false.
+ processingBetweenOperator(expr, searchConditions, nodeOutputs);
+ return expr;
+ } else if (genericUDF instanceof GenericUDFIn) {
+ // In case of not in operator, in operator exist as child of not operator.
+ processingInOperator(expr, searchConditions, false, nodeOutputs);
+ return expr;
+ } else if (genericUDF instanceof GenericUDFOPNot &&
+ ((ExprNodeGenericFuncDesc) expr.getChildren().get(0)).getGenericUDF()
+ instanceof GenericUDFIn) {
+ // In case of not in operator, in operator exist as child of not operator.
+ processingInOperator((ExprNodeGenericFuncDesc) expr.getChildren().get(0),
+ searchConditions, true, ((ExprNodeGenericFuncDesc) nodeOutputs[0])
+ .getChildren().toArray());
+ return expr;
+ } else if (genericUDF instanceof GenericUDFOPNull) {
+ processingNullOperator(expr, searchConditions, nodeOutputs);
+ return expr;
+ } else if (genericUDF instanceof GenericUDFOPNotNull) {
+ processingNotNullOperator(expr, searchConditions, nodeOutputs);
+ return expr;
+ } else {
+ return expr;
+ }
+ }
+ ExprNodeDesc expr1 = (ExprNodeDesc) nodeOutputs[0];
+ ExprNodeDesc expr2 = (ExprNodeDesc) nodeOutputs[1];
+ // We may need to peel off the GenericUDFBridge that is added by CBO or
+ // user
+ if (expr1.getTypeInfo().equals(expr2.getTypeInfo())) {
+ expr1 = getColumnExpr(expr1);
+ expr2 = getColumnExpr(expr2);
+ }
+
+ ExprNodeDesc[] extracted = ExprNodeDescUtils.extractComparePair(expr1, expr2);
+ if (extracted == null || (extracted.length > 2 && !acceptsFields)) {
+ return expr;
+ }
+
+ ExprNodeColumnDesc columnDesc;
+ ExprNodeConstantDesc constantDesc;
+ if (extracted[0] instanceof ExprNodeConstantDesc) {
+ genericUDF = genericUDF.flip();
+ columnDesc = (ExprNodeColumnDesc) extracted[1];
+ constantDesc = (ExprNodeConstantDesc) extracted[0];
+ } else {
+ columnDesc = (ExprNodeColumnDesc) extracted[0];
+ constantDesc = (ExprNodeConstantDesc) extracted[1];
+ }
+
+ Set<String> allowed = columnToUDFs.get(columnDesc.getColumn());
+ if (allowed == null) {
+ return expr;
+ }
+
+ String udfName = genericUDF.getUdfName();
+ if (!allowed.contains(genericUDF.getUdfName())) {
+ return expr;
+ }
+
+ String[] fields = null;
+ if (extracted.length > 2) {
+ ExprNodeFieldDesc fieldDesc = (ExprNodeFieldDesc) extracted[2];
+ if (!isValidField(fieldDesc)) {
+ return expr;
+ }
+ fields = ExprNodeDescUtils.extractFields(fieldDesc);
+ }
+
+ // We also need to update the expr so that the index query can be
+ // generated.
+ // Note that, hive does not support UDFToDouble etc in the query text.
+ List<ExprNodeDesc> list = new ArrayList<ExprNodeDesc>();
+ list.add(expr1);
+ list.add(expr2);
+ expr = new ExprNodeGenericFuncDesc(expr.getTypeInfo(), expr.getGenericUDF(), list);
+
+ searchConditions.add(new IndexSearchCondition(columnDesc, udfName, constantDesc, expr,
+ fields));
+
+ // we converted the expression to a search condition, so
+ // remove it from the residual predicate
+ return fields == null ? null : expr;
+ }
+
+ private boolean isValidField(ExprNodeFieldDesc field) {
+ return fieldValidator == null || fieldValidator.validate(field);
+ }
+
+ /**
+ * Translates search conditions back to ExprNodeDesc form (as a left-deep
+ * conjunction).
+ *
+ * @param searchConditions (typically produced by analyzePredicate)
+ * @return ExprNodeGenericFuncDesc form of search conditions
+ */
+ public ExprNodeGenericFuncDesc translateSearchConditions(List<IndexSearchCondition>
+ searchConditions) {
+
+ ExprNodeGenericFuncDesc expr = null;
+
+ for (IndexSearchCondition searchCondition : searchConditions) {
+ if (expr == null) {
+ expr = searchCondition.getComparisonExpr();
+ continue;
+ }
+
+ List<ExprNodeDesc> children = new ArrayList<ExprNodeDesc>();
+ children.add(expr);
+ children.add(searchCondition.getComparisonExpr());
+ expr = new ExprNodeGenericFuncDesc(TypeInfoFactory.booleanTypeInfo, FunctionRegistry
+ .getGenericUDFForAnd(), children);
+ }
+
+ return expr;
+ }
+
+ public void setAcceptsFields(boolean acceptsFields) {
+ this.acceptsFields = acceptsFields;
+ }
+
+ public static interface FieldValidator {
+ boolean validate(ExprNodeFieldDesc exprNodeDesc);
+ }
+
+ public static IndexPredicateAnalyzer createAnalyzer(boolean equalOnly) {
+ IndexPredicateAnalyzer analyzer = new IndexPredicateAnalyzer();
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqual");
+
+ if (equalOnly) {
+ return analyzer;
+ }
+
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic" +
+ ".GenericUDFOPEqualOrGreaterThan");
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic" +
+ ".GenericUDFOPEqualOrLessThan");
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPLessThan");
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPGreaterThan");
+
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotEqual");
+ // apply !=
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFBetween");
+ // apply (Not) Between
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFIn"); //
+ // apply (Not) In
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFIn"); //
+ // apply In
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNull");
+ // apply Null
+ analyzer.addComparisonOp("org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPNotNull");
+ // apply Not Null
+
+ return analyzer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/537b90be/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexSearchCondition.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexSearchCondition.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexSearchCondition.java
new file mode 100644
index 0000000..ebd978c
--- /dev/null
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/IndexSearchCondition.java
@@ -0,0 +1,143 @@
+/*
+ * 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.phoenix.hive.ql.index;
+
+import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
+
+/**
+ * IndexSearchCondition represents an individual search condition found by
+ * {@link IndexPredicateAnalyzer}.
+ *
+ */
+public class IndexSearchCondition {
+ private ExprNodeColumnDesc columnDesc;
+ private String comparisonOp;
+ private ExprNodeConstantDesc constantDesc;
+ private ExprNodeGenericFuncDesc comparisonExpr;
+
+ private String[] fields;
+
+ // Support (Not) Between/(Not) In Operator
+ private ExprNodeConstantDesc[] multiConstants;
+ private boolean isNot;
+
+ public IndexSearchCondition(ExprNodeColumnDesc columnDesc, String comparisonOp,
+ ExprNodeConstantDesc[] multiConstants, ExprNodeGenericFuncDesc
+ comparisonExpr, boolean isNot) {
+ this(columnDesc, comparisonOp, multiConstants, comparisonExpr, null, isNot);
+ }
+
+ public IndexSearchCondition(ExprNodeColumnDesc columnDesc, String comparisonOp,
+ ExprNodeConstantDesc[] multiConstants, ExprNodeGenericFuncDesc
+ comparisonExpr, String[] fields, boolean isNot) {
+ this.columnDesc = columnDesc;
+ this.comparisonOp = comparisonOp;
+ this.multiConstants = multiConstants;
+ this.comparisonExpr = comparisonExpr;
+ this.fields = fields;
+ this.isNot = isNot;
+ }
+
+ public ExprNodeConstantDesc[] getConstantDescs() {
+ return multiConstants;
+ }
+
+ public ExprNodeConstantDesc getConstantDesc(int index) {
+ return multiConstants[index];
+ }
+
+ public boolean isNot() {
+ return isNot;
+ }
+ //////////////////////////////////////////////////////////////////////////////
+
+ public IndexSearchCondition(ExprNodeColumnDesc columnDesc, String comparisonOp,
+ ExprNodeConstantDesc constantDesc, ExprNodeGenericFuncDesc
+ comparisonExpr) {
+ this(columnDesc, comparisonOp, constantDesc, comparisonExpr, null);
+ }
+
+ /**
+ * Constructs a search condition, which takes the form
+ * <p>
+ * <pre>
+ * column-ref comparison-op constant-value
+ * </pre>
+ * <p>
+ * .
+ *
+ * @param columnDesc column being compared
+ * @param comparisonOp comparison operator, e.g. "=" (taken from
+ * GenericUDFBridge.getUdfName())
+ * @param constantDesc constant value to search for
+ * @param comparisonExpr the original comparison expression
+ */
+ public IndexSearchCondition(ExprNodeColumnDesc columnDesc, String comparisonOp,
+ ExprNodeConstantDesc constantDesc, ExprNodeGenericFuncDesc
+ comparisonExpr, String[] fields) {
+
+ this.columnDesc = columnDesc;
+ this.comparisonOp = comparisonOp;
+ this.constantDesc = constantDesc;
+ this.comparisonExpr = comparisonExpr;
+ this.fields = fields;
+ }
+
+ public void setColumnDesc(ExprNodeColumnDesc columnDesc) {
+ this.columnDesc = columnDesc;
+ }
+
+ public ExprNodeColumnDesc getColumnDesc() {
+ return columnDesc;
+ }
+
+ public void setComparisonOp(String comparisonOp) {
+ this.comparisonOp = comparisonOp;
+ }
+
+ public String getComparisonOp() {
+ return comparisonOp;
+ }
+
+ public void setConstantDesc(ExprNodeConstantDesc constantDesc) {
+ this.constantDesc = constantDesc;
+ }
+
+ public ExprNodeConstantDesc getConstantDesc() {
+ return constantDesc;
+ }
+
+ public void setComparisonExpr(ExprNodeGenericFuncDesc comparisonExpr) {
+ this.comparisonExpr = comparisonExpr;
+ }
+
+ public ExprNodeGenericFuncDesc getComparisonExpr() {
+ return comparisonExpr;
+ }
+
+ public String[] getFields() {
+ return fields;
+ }
+
+ @Override
+ public String toString() {
+ return comparisonExpr.getExprString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/537b90be/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/PredicateAnalyzerFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/PredicateAnalyzerFactory.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/PredicateAnalyzerFactory.java
new file mode 100644
index 0000000..b6903b9
--- /dev/null
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/ql/index/PredicateAnalyzerFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.phoenix.hive.ql.index;
+
+import org.apache.phoenix.hive.ql.index.IndexPredicateAnalyzer.FieldValidator;
+
+import java.util.List;
+
+public class PredicateAnalyzerFactory {
+ public static IndexPredicateAnalyzer createPredicateAnalyzer(List<String> ppdColumnList,
+ FieldValidator fieldValdator) {
+ // Create analyzer for conditions =, <, <=, >, >=
+ IndexPredicateAnalyzer analyzer = IndexPredicateAnalyzer.createAnalyzer(false);
+
+ for (String columnName : ppdColumnList) {
+ analyzer.allowColumnName(columnName);
+ }
+
+ analyzer.setAcceptsFields(true);
+ analyzer.setFieldValidator(fieldValdator);
+
+ return analyzer;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/537b90be/phoenix-hive/src/main/java/org/apache/phoenix/hive/query/PhoenixQueryBuilder.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/query/PhoenixQueryBuilder.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/query/PhoenixQueryBuilder.java
new file mode 100644
index 0000000..8e3a972
--- /dev/null
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/query/PhoenixQueryBuilder.java
@@ -0,0 +1,760 @@
+/*
+ * 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.phoenix.hive.query;
+
+import com.google.common.base.CharMatcher;
+import com.google.common.base.Joiner;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.phoenix.hive.constants.PhoenixStorageHandlerConstants;
+import org.apache.phoenix.hive.ql.index.IndexSearchCondition;
+import org.apache.phoenix.hive.util.PhoenixStorageHandlerUtil;
+import org.apache.phoenix.hive.util.PhoenixUtil;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Query builder. Produces a query depending on the colummn list and conditions
+ */
+
+public class PhoenixQueryBuilder {
+
+ private static final Log LOG = LogFactory.getLog(PhoenixQueryBuilder.class);
+
+ private static final String QUERY_TEMPLATE = "select $HINT$ $COLUMN_LIST$ from $TABLE_NAME$";
+
+ private static final PhoenixQueryBuilder QUERY_BUILDER = new PhoenixQueryBuilder();
+
+ private PhoenixQueryBuilder() {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("PhoenixQueryBuilder created");
+ }
+ }
+
+ public static PhoenixQueryBuilder getInstance() {
+ return QUERY_BUILDER;
+ }
+
+ private void addConditionColumnToReadColumn(List<String> readColumnList, List<String>
+ conditionColumnList) {
+ if (readColumnList.isEmpty()) {
+ return;
+ }
+
+ for (String conditionColumn : conditionColumnList) {
+ if (!readColumnList.contains(conditionColumn)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Condition column " + conditionColumn + " does not exist in " +
+ "read-columns.");
+ }
+
+ readColumnList.add(conditionColumn);
+ }
+ }
+ }
+
+ private String makeQueryString(JobConf jobConf, String tableName, List<String>
+ readColumnList, String whereClause, String queryTemplate, String hints, Map<String,
+ String> columnTypeMap) throws IOException {
+ StringBuilder sql = new StringBuilder();
+ List<String> conditionColumnList = buildWhereClause(jobConf, sql, whereClause,
+ columnTypeMap);
+
+ if (conditionColumnList.size() > 0) {
+ addConditionColumnToReadColumn(readColumnList, conditionColumnList);
+ sql.insert(0, queryTemplate.replace("$HINT$", hints).replace("$COLUMN_LIST$",
+ getSelectColumns(jobConf, tableName, readColumnList)).replace("$TABLE_NAME$",
+ tableName));
+ } else {
+ sql.append(queryTemplate.replace("$HINT$", hints).replace("$COLUMN_LIST$",
+ getSelectColumns(jobConf, tableName, readColumnList)).replace("$TABLE_NAME$",
+ tableName));
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Input query : " + sql.toString());
+ }
+
+ return sql.toString();
+ }
+
+ private String makeQueryString(JobConf jobConf, String tableName, List<String>
+ readColumnList, List<IndexSearchCondition> searchConditions, String queryTemplate,
+ String hints) throws IOException {
+ StringBuilder sql = new StringBuilder();
+ List<String> conditionColumnList = buildWhereClause(jobConf, sql, searchConditions);
+
+ if (conditionColumnList.size() > 0) {
+ addConditionColumnToReadColumn(readColumnList, conditionColumnList);
+ sql.insert(0, queryTemplate.replace("$HINT$", hints).replace("$COLUMN_LIST$",
+ getSelectColumns(jobConf, tableName, readColumnList)).replace("$TABLE_NAME$",
+ tableName));
+ } else {
+ sql.append(queryTemplate.replace("$HINT$", hints).replace("$COLUMN_LIST$",
+ getSelectColumns(jobConf, tableName, readColumnList)).replace("$TABLE_NAME$",
+ tableName));
+ }
+
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Input query : " + sql.toString());
+ }
+
+ return sql.toString();
+ }
+
+ private String getSelectColumns(JobConf jobConf, String tableName, List<String>
+ readColumnList) throws IOException {
+ String selectColumns = Joiner.on(PhoenixStorageHandlerConstants.COMMA).join(readColumnList);
+
+ if (PhoenixStorageHandlerConstants.EMPTY_STRING.equals(selectColumns)) {
+ selectColumns = "*";
+ } else {
+ if (PhoenixStorageHandlerUtil.isTransactionalTable(jobConf)) {
+ List<String> pkColumnList = PhoenixUtil.getPrimaryKeyColumnList(jobConf, tableName);
+ StringBuilder pkColumns = new StringBuilder();
+
+ for (String pkColumn : pkColumnList) {
+ String pkColumnName = pkColumn.toLowerCase();
+
+ if (!readColumnList.contains(pkColumnName)) {
+ pkColumns.append(pkColumnName).append(PhoenixStorageHandlerConstants.COMMA);
+ }
+ }
+
+ selectColumns = pkColumns.toString() + selectColumns;
+ }
+ }
+
+ return selectColumns;
+ }
+
+ public String buildQuery(JobConf jobConf, String tableName, List<String> readColumnList,
+ String whereClause, Map<String, String> columnTypeMap) throws
+ IOException {
+ String hints = getHint(jobConf, tableName);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Building query with columns : " + readColumnList + " table name : " +
+ tableName + " with where conditions : " + whereClause + " hints : " + hints);
+ }
+
+ return makeQueryString(jobConf, tableName, Lists.newArrayList(readColumnList),
+ whereClause, QUERY_TEMPLATE, hints, columnTypeMap);
+ }
+
+ public String buildQuery(JobConf jobConf, String tableName, List<String> readColumnList,
+ List<IndexSearchCondition> searchConditions) throws IOException {
+ String hints = getHint(jobConf, tableName);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Building query with columns : " + readColumnList + " table name : " +
+ tableName + " search conditions : " + searchConditions + " hints : " + hints);
+ }
+
+ return makeQueryString(jobConf, tableName, Lists.newArrayList(readColumnList),
+ searchConditions, QUERY_TEMPLATE, hints);
+ }
+
+ private String getHint(JobConf jobConf, String tableName) {
+ StringBuilder hints = new StringBuilder("/*+ ");
+ if (!jobConf.getBoolean(PhoenixStorageHandlerConstants.HBASE_SCAN_CACHEBLOCKS, Boolean
+ .FALSE)) {
+ hints.append("NO_CACHE ");
+ }
+
+ String queryHint = jobConf.get(tableName + PhoenixStorageHandlerConstants
+ .PHOENIX_TABLE_QUERY_HINT);
+ if (queryHint != null) {
+ hints.append(queryHint);
+ }
+ hints.append(" */");
+
+ return hints.toString();
+ }
+
+ private List<String> buildWhereClause(JobConf jobConf, StringBuilder sql, String whereClause,
+ Map<String, String> columnTypeMap) throws IOException {
+ if (whereClause == null || whereClause.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<String> conditionColumnList = Lists.newArrayList();
+ sql.append(" where ");
+
+ whereClause = StringUtils.replaceEach(whereClause, new String[]{"UDFToString"}, new
+ String[]{"to_char"});
+
+ for (String columnName : columnTypeMap.keySet()) {
+ if (whereClause.contains(columnName)) {
+ conditionColumnList.add(columnName);
+
+ if (PhoenixStorageHandlerConstants.DATE_TYPE.equals(columnTypeMap.get(columnName)
+ )) {
+ whereClause = applyDateFunctionUsingRegex(whereClause, columnName);
+ } else if (PhoenixStorageHandlerConstants.TIMESTAMP_TYPE.equals(columnTypeMap.get
+ (columnName))) {
+ whereClause = applyTimestampFunctionUsingRegex(whereClause, columnName);
+ }
+ }
+ }
+
+ sql.append(whereClause);
+
+ return conditionColumnList;
+ }
+
+ private String applyDateFunctionUsingRegex(String whereClause, String columnName) {
+ whereClause = applyFunctionForCommonOperator(whereClause, columnName, true);
+ whereClause = applyFunctionForBetweenOperator(whereClause, columnName, true);
+ whereClause = applyFunctionForInOperator(whereClause, columnName, true);
+
+ return whereClause;
+ }
+
+ private String applyTimestampFunctionUsingRegex(String whereClause, String columnName) {
+ whereClause = applyFunctionForCommonOperator(whereClause, columnName, false);
+ whereClause = applyFunctionForBetweenOperator(whereClause, columnName, false);
+ whereClause = applyFunctionForInOperator(whereClause, columnName, false);
+
+ return whereClause;
+ }
+
+ private String applyFunctionForCommonOperator(String whereClause, String columnName, boolean
+ isDate) {
+ String targetPattern = isDate ? PhoenixStorageHandlerConstants.DATE_PATTERN :
+ PhoenixStorageHandlerConstants.TIMESTAMP_PATTERN;
+ String pattern = StringUtils.replaceEach(PhoenixStorageHandlerConstants
+ .COMMON_OPERATOR_PATTERN,
+ new String[]{PhoenixStorageHandlerConstants.COLUMNE_MARKER,
+ PhoenixStorageHandlerConstants.PATERN_MARKER}, new String[]{columnName,
+ targetPattern});
+
+ Matcher matcher = Pattern.compile(pattern).matcher(whereClause);
+
+ while (matcher.find()) {
+ String token = matcher.group(1);
+ String datePart = matcher.group(3);
+
+ String convertString = token.replace(datePart, applyFunction(isDate ?
+ PhoenixStorageHandlerConstants.DATE_FUNCTION_TEMPLETE :
+ PhoenixStorageHandlerConstants.TIMESTAMP_FUNCTION_TEMPLATE, datePart));
+ whereClause = whereClause.replaceAll(StringUtils.replaceEach(token, new String[]{"(",
+ ")"}, new String[]{"\\(", "\\)"}), convertString);
+ }
+
+ return whereClause;
+ }
+
+ private String applyFunctionForBetweenOperator(String whereClause, String columnName, boolean
+ isDate) {
+ String targetPattern = isDate ? PhoenixStorageHandlerConstants.DATE_PATTERN :
+ PhoenixStorageHandlerConstants.TIMESTAMP_PATTERN;
+ String pattern = StringUtils.replaceEach(PhoenixStorageHandlerConstants
+ .BETWEEN_OPERATOR_PATTERN,
+ new String[]{PhoenixStorageHandlerConstants.COLUMNE_MARKER,
+ PhoenixStorageHandlerConstants.PATERN_MARKER}, new String[]{columnName,
+ targetPattern});
+
+ Matcher matcher = Pattern.compile(pattern).matcher(whereClause);
+
+ while (matcher.find()) {
+ String token = matcher.group(1);
+ boolean isNot = matcher.group(2) == null ? false : true;
+ String fromDate = matcher.group(3);
+ String toDate = matcher.group(4);
+
+ String convertString = StringUtils.replaceEach(token, new String[]{fromDate, toDate},
+ new String[]{applyFunction(isDate ? PhoenixStorageHandlerConstants
+ .DATE_FUNCTION_TEMPLETE : PhoenixStorageHandlerConstants
+ .TIMESTAMP_FUNCTION_TEMPLATE, fromDate),
+ applyFunction(isDate ? PhoenixStorageHandlerConstants
+ .DATE_FUNCTION_TEMPLETE : PhoenixStorageHandlerConstants
+ .TIMESTAMP_FUNCTION_TEMPLATE, toDate)});
+
+ whereClause = whereClause.replaceAll(pattern, convertString);
+ }
+
+ return whereClause;
+ }
+
+ private String applyFunctionForInOperator(String whereClause, String columnName, boolean
+ isDate) {
+ String targetPattern = isDate ? PhoenixStorageHandlerConstants.DATE_PATTERN :
+ PhoenixStorageHandlerConstants.TIMESTAMP_PATTERN;
+ String pattern = StringUtils.replaceEach(PhoenixStorageHandlerConstants.IN_OPERATOR_PATTERN,
+ new String[]{PhoenixStorageHandlerConstants.COLUMNE_MARKER,
+ PhoenixStorageHandlerConstants.PATERN_MARKER}, new String[]{columnName,
+ targetPattern});
+ String itemPattern = "(" + targetPattern + ")";
+
+ Matcher matcher = Pattern.compile(pattern).matcher(whereClause);
+
+ while (matcher.find()) {
+ String token = matcher.group(1);
+ Matcher itemMatcher = Pattern.compile(itemPattern).matcher(token);
+ while (itemMatcher.find()) {
+ String item = itemMatcher.group(1);
+
+ token = token.replace(item, applyFunction(isDate ? PhoenixStorageHandlerConstants
+ .DATE_FUNCTION_TEMPLETE : PhoenixStorageHandlerConstants
+ .TIMESTAMP_FUNCTION_TEMPLATE, item));
+ }
+
+ whereClause = whereClause.replaceAll(pattern, token);
+ }
+
+ return whereClause;
+ }
+
+ /**
+ * replace value to specific part of pattern.
+ * if pattern is to_date($value$) and value is '2016-01-15'. then return to_date('2016-01-15').
+ * if pattern is cast($value$ as date) and value is '2016-01-15'. then return cast
+ * ('2016-01-15' as date).
+ */
+ private String applyFunction(String pattern, String value) {
+ if (!value.startsWith(PhoenixStorageHandlerConstants.QUOTATION_MARK)) {
+ value = PhoenixStorageHandlerConstants.QUOTATION_MARK + value +
+ PhoenixStorageHandlerConstants.QUOTATION_MARK;
+ }
+
+ return pattern.replace(PhoenixStorageHandlerConstants.FUNCTION_VALUE_MARKER, value);
+ }
+
+ private String getCompareValueForDateAndTimestampFunction(String compareValue) {
+ if (compareValue.startsWith(PhoenixStorageHandlerConstants.QUOTATION_MARK)) {
+ return compareValue;
+ } else {
+ return PhoenixStorageHandlerConstants.QUOTATION_MARK + compareValue +
+ PhoenixStorageHandlerConstants.QUOTATION_MARK;
+ }
+ }
+
+ private String applyDateFunction(String whereClause, String columnName) {
+ StringBuilder whereCondition = new StringBuilder();
+ for (Iterator<String> iterator = Splitter.on(CharMatcher.WHITESPACE).omitEmptyStrings()
+ .split(whereClause).iterator(); iterator.hasNext(); whereCondition.append
+ (PhoenixStorageHandlerConstants.SPACE)) {
+ String token = iterator.next();
+ if (isMyCondition(columnName, token)) {
+ whereCondition.append(token);
+
+ String comparator = iterator.next();
+ whereCondition.append(PhoenixStorageHandlerConstants.SPACE);
+ whereCondition.append(comparator).append(PhoenixStorageHandlerConstants.SPACE);
+ if (PhoenixStorageHandlerConstants.BETWEEN_COMPARATOR.equalsIgnoreCase
+ (comparator)) {
+ whereCondition.append("to_date(").append
+ (getCompareValueForDateAndTimestampFunction(iterator.next())).append
+ (") ").append(iterator.next()).append(PhoenixStorageHandlerConstants
+ .SPACE)
+ .append("to_date(");
+
+ String toCompareValue = iterator.next();
+ if (toCompareValue.endsWith(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ int rightBracketIndex = toCompareValue.indexOf
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (toCompareValue.substring(0, rightBracketIndex))).append
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET).append
+ (toCompareValue.substring(rightBracketIndex));
+ } else {
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (toCompareValue)).append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET);
+ }
+ } else if (PhoenixStorageHandlerConstants.IN_COMPARATOR.equalsIgnoreCase
+ (comparator)) {
+ while (iterator.hasNext()) {
+ String aToken = iterator.next();
+ if (aToken.equals(PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET) ||
+ aToken.equals(PhoenixStorageHandlerConstants.COMMA)) {
+ whereCondition.append(aToken);
+ } else if (aToken.equals(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ whereCondition.append(aToken);
+ break;
+ } else if (aToken.endsWith(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ int bracketIndex = aToken.indexOf(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET);
+ whereCondition.append("to_date(").append
+ (getCompareValueForDateAndTimestampFunction(aToken.substring
+ (0, bracketIndex))).append
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET).append
+ (aToken.substring(bracketIndex));
+ break;
+ } else if (aToken.endsWith(PhoenixStorageHandlerConstants.COMMA)) {
+ if (aToken.startsWith(PhoenixStorageHandlerConstants
+ .LEFT_ROUND_BRACKET)) {
+ int bracketIndex = aToken.lastIndexOf
+ (PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET);
+ whereCondition.append(aToken.substring(0, bracketIndex + 1))
+ .append("to_date(").append
+ (getCompareValueForDateAndTimestampFunction(aToken
+ .substring(bracketIndex + 1, aToken.length() - 1)
+ )).append("),");
+ } else {
+ whereCondition.append("to_date(").append
+ (getCompareValueForDateAndTimestampFunction(aToken
+ .substring(0, aToken.length() - 1))).append("),");
+ }
+ }
+
+ whereCondition.append(PhoenixStorageHandlerConstants.SPACE);
+ }
+ } else if (PhoenixStorageHandlerConstants.COMMON_COMPARATOR.contains(comparator)) {
+ String compareValue = getCompareValueForDateAndTimestampFunction(iterator
+ .next());
+ whereCondition.append("to_date(");
+ if (compareValue.endsWith(PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET)) {
+ int rightBracketIndex = compareValue.indexOf
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (compareValue.substring(0, rightBracketIndex))).append
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET).append
+ (compareValue.substring(rightBracketIndex));
+ } else {
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (compareValue)).append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET);
+ }
+ }
+ } else {
+ whereCondition.append(token);
+ }
+ }
+
+ return whereCondition.toString();
+ }
+
+ // Assume timestamp value is yyyy-MM-dd HH:mm:ss.SSS
+ private String applyTimestampFunction(String whereClause, String columnName) {
+ StringBuilder whereCondition = new StringBuilder();
+ for (Iterator<String> iterator = Splitter.on(CharMatcher.WHITESPACE).omitEmptyStrings()
+ .split(whereClause).iterator(); iterator.hasNext(); whereCondition.append
+ (PhoenixStorageHandlerConstants.SPACE)) {
+ String token = iterator.next();
+ if (isMyCondition(columnName, token)) {
+ whereCondition.append(token);
+
+ String comparator = iterator.next();
+ whereCondition.append(PhoenixStorageHandlerConstants.SPACE);
+ whereCondition.append(comparator).append(PhoenixStorageHandlerConstants.SPACE);
+ if (PhoenixStorageHandlerConstants.BETWEEN_COMPARATOR.equalsIgnoreCase
+ (comparator)) {
+ String fromCompareValue = iterator.next() + PhoenixStorageHandlerConstants
+ .SPACE + iterator.next();
+ whereCondition.append("to_timestamp(").append
+ (getCompareValueForDateAndTimestampFunction(fromCompareValue)).append
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append(PhoenixStorageHandlerConstants.SPACE).append(iterator
+ .next()).append(PhoenixStorageHandlerConstants.SPACE);
+ whereCondition.append("to_timestamp(");
+
+ String toCompareValue = iterator.next() + PhoenixStorageHandlerConstants
+ .SPACE + iterator.next();
+ if (toCompareValue.endsWith(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ int rightBracketIndex = toCompareValue.indexOf
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (toCompareValue.substring(0, rightBracketIndex))).append
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET).append
+ (toCompareValue.substring(rightBracketIndex));
+ } else {
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (toCompareValue)).append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET);
+ }
+ } else if (PhoenixStorageHandlerConstants.IN_COMPARATOR.equalsIgnoreCase
+ (comparator)) {
+ while (iterator.hasNext()) {
+ String aToken = iterator.next();
+ if (aToken.equals(PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET) ||
+ aToken.equals(PhoenixStorageHandlerConstants.COMMA)) {
+ whereCondition.append(aToken);
+ } else if (aToken.equals(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ whereCondition.append(aToken);
+ break;
+ } else {
+ String compareValue = aToken + PhoenixStorageHandlerConstants.SPACE +
+ iterator.next();
+
+ if (compareValue.startsWith(PhoenixStorageHandlerConstants
+ .LEFT_ROUND_BRACKET)) {
+ int leftBracketIndex = compareValue.lastIndexOf
+ (PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET);
+ whereCondition.append(compareValue.substring(0, leftBracketIndex
+ + 1)).append("to_timestamp(");
+
+ if (compareValue.endsWith(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ int rightBracketIndex = compareValue.indexOf
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append
+ (getCompareValueForDateAndTimestampFunction
+ (compareValue.substring(leftBracketIndex + 1,
+ rightBracketIndex)))
+ .append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET).append(compareValue
+ .substring(rightBracketIndex));
+ } else if (compareValue.endsWith(PhoenixStorageHandlerConstants
+ .COMMA)) {
+ whereCondition.append
+ (getCompareValueForDateAndTimestampFunction
+ (compareValue.substring(leftBracketIndex + 1,
+ compareValue.length() - 1)))
+ .append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET).append
+ (PhoenixStorageHandlerConstants.COMMA);
+ } else {
+ whereCondition.append
+ (getCompareValueForDateAndTimestampFunction
+ (compareValue.substring(leftBracketIndex + 1)
+ )).append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET);
+ }
+ } else if (compareValue.endsWith(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ int rightBracketIndex = compareValue.indexOf
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append("to_timestamp(").append
+ (getCompareValueForDateAndTimestampFunction(compareValue
+ .substring(0, rightBracketIndex)))
+ .append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET).append(compareValue
+ .substring(rightBracketIndex));
+ break;
+ } else if (compareValue.endsWith(PhoenixStorageHandlerConstants
+ .COMMA)) {
+ whereCondition.append("to_timestamp(").append
+ (getCompareValueForDateAndTimestampFunction(compareValue
+ .substring(0, compareValue.length() - 1))).append
+ ("),");
+ }
+ }
+
+ whereCondition.append(PhoenixStorageHandlerConstants.SPACE);
+ }
+ } else if (PhoenixStorageHandlerConstants.COMMON_COMPARATOR.contains(comparator)) {
+ String timestampValue = iterator.next() + PhoenixStorageHandlerConstants
+ .SPACE + iterator.next();
+ whereCondition.append("to_timestamp(");
+ if (timestampValue.endsWith(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)) {
+ int rightBracketIndex = timestampValue.indexOf
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET);
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (timestampValue.substring(0, rightBracketIndex))).append
+ (PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET).append
+ (timestampValue.substring(rightBracketIndex));
+ } else {
+ whereCondition.append(getCompareValueForDateAndTimestampFunction
+ (timestampValue)).append(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET);
+ }
+ }
+ } else {
+ whereCondition.append(token);
+ }
+ }
+
+ return whereCondition.toString();
+ }
+
+ private boolean isMyCondition(String columnName, String token) {
+ boolean itsMine = false;
+
+ if (columnName.equals(token)) {
+ itsMine = true;
+ } else if (token.startsWith(PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET) && token
+ .substring(token.lastIndexOf(PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET) +
+ 1).equals(columnName)) {
+ itsMine = true;
+ } else if (token.startsWith(PhoenixStorageHandlerConstants.LEFT_ROUND_BRACKET) && token
+ .endsWith(PhoenixStorageHandlerConstants.RIGHT_ROUND_BRACKET)
+ && token.substring(token.lastIndexOf(PhoenixStorageHandlerConstants
+ .LEFT_ROUND_BRACKET) + 1, token.indexOf(PhoenixStorageHandlerConstants
+ .RIGHT_ROUND_BRACKET)).equals(columnName)) {
+ itsMine = true;
+ }
+
+ return itsMine;
+ }
+
+ protected List<String> buildWhereClause(JobConf jobConf, StringBuilder sql,
+ List<IndexSearchCondition> searchConditions) throws
+ IOException {
+ if (searchConditions == null || searchConditions.size() == 0) {
+ return Collections.emptyList();
+ }
+
+ List<String> conditionColumnList = Lists.newArrayList();
+ sql.append(" where ");
+
+ boolean firstCondition = true;
+ for (IndexSearchCondition condition : searchConditions) {
+ String comparisonOp = condition.getComparisonOp();
+
+ if (comparisonOp.endsWith("GenericUDFBetween") || comparisonOp.endsWith
+ ("GenericUDFIn")) {
+ if (condition.getConstantDescs() == null) {
+ continue;
+ }
+ } else if (comparisonOp.endsWith("GenericUDFOPNull") || comparisonOp.endsWith
+ ("GenericUDFOPNotNull")) {
+ // keep going
+ } else if (comparisonOp.endsWith("GenericUDFOPEqual")) {
+ // keep going
+ } else {
+ if (condition.getConstantDesc().getValue() == null) {
+ continue;
+ }
+ }
+
+ if (!firstCondition) {
+ sql.append(" and ");
+ } else {
+ firstCondition = false;
+ }
+
+ String columnName = condition.getColumnDesc().getColumn();
+ String typeName = condition.getColumnDesc().getTypeString();
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(columnName + " has condition: " + condition);
+ }
+
+ conditionColumnList.add(columnName);
+ sql.append(columnName);
+
+ String[] constantValues = PhoenixStorageHandlerUtil.getConstantValues(condition,
+ comparisonOp);
+
+ if (comparisonOp.endsWith("UDFOPEqual")) { // column = 1
+ appendCondition(sql, " = ", typeName, constantValues[0]);
+ } else if (comparisonOp.endsWith("UDFOPEqualOrGreaterThan")) { // column >= 1
+ appendCondition(sql, " >= ", typeName, constantValues[0]);
+ } else if (comparisonOp.endsWith("UDFOPGreaterThan")) { // column > 1
+ appendCondition(sql, " > ", typeName, constantValues[0]);
+ } else if (comparisonOp.endsWith("UDFOPEqualOrLessThan")) { // column <= 1
+ appendCondition(sql, " <= ", typeName, constantValues[0]);
+ } else if (comparisonOp.endsWith("UDFOPLessThan")) { // column < 1
+ appendCondition(sql, " < ", typeName, constantValues[0]);
+ } else if (comparisonOp.endsWith("UDFOPNotEqual")) { // column != 1
+ appendCondition(sql, " != ", typeName, constantValues[0]);
+ } else if (comparisonOp.endsWith("GenericUDFBetween")) {
+ appendBetweenCondition(jobConf, sql, condition.isNot(), typeName, constantValues);
+ } else if (comparisonOp.endsWith("GenericUDFIn")) {
+ appendInCondition(sql, condition.isNot(), typeName, constantValues);
+ } else if (comparisonOp.endsWith("GenericUDFOPNull")) {
+ sql.append(" is null ");
+ } else if (comparisonOp.endsWith("GenericUDFOPNotNull")) {
+ sql.append(" is not null ");
+ }
+ }
+
+ return conditionColumnList;
+ }
+
+ protected void appendCondition(StringBuilder sql, String comparisonOp, String typeName,
+ String conditionValue) {
+ if (serdeConstants.STRING_TYPE_NAME.equals(typeName)) {
+ sql.append(comparisonOp).append("'").append(conditionValue).append("'");
+ } else if (serdeConstants.DATE_TYPE_NAME.equals(typeName)) {
+ sql.append(comparisonOp).append("to_date('").append(conditionValue).append("')");
+ } else if (serdeConstants.TIMESTAMP_TYPE_NAME.equals(typeName)) {
+ sql.append(comparisonOp).append("to_timestamp('").append(conditionValue).append("')");
+ } else {
+ sql.append(comparisonOp).append(conditionValue);
+ }
+ }
+
+ protected void appendBetweenCondition(JobConf jobConf, StringBuilder sql, boolean isNot,
+ String typeName, String[] conditionValues) throws
+ IOException {
+ if (isNot) {
+ sql.append(" not between ");
+ } else {
+ sql.append(" between ");
+ }
+
+ try {
+ Arrays.sort(PhoenixStorageHandlerUtil.toTypedValues(jobConf, typeName,
+ conditionValues));
+
+ if (serdeConstants.STRING_TYPE_NAME.equals(typeName)) {
+ sql.append("'").append(conditionValues[0]).append("'").append(" and ").append
+ ("'").append(conditionValues[1]).append("'");
+ } else if (serdeConstants.DATE_TYPE_NAME.equals(typeName)) {
+ sql.append("to_date('").append(conditionValues[0]).append("')").append(" and ")
+ .append("to_date('").append(conditionValues[1]).append("')");
+ } else if (serdeConstants.TIMESTAMP_TYPE_NAME.equals(typeName)) {
+ sql.append("to_timestamp('").append(conditionValues[0]).append("')").append(" and" +
+ " ").append("to_timestamp('").append(conditionValues[1]).append("')");
+ } else {
+ sql.append(conditionValues[0]).append(" and ").append(conditionValues[1]);
+ }
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+
+ protected void appendInCondition(StringBuilder sql, boolean isNot, String typeName, String[]
+ conditionValues) {
+ if (isNot) {
+ sql.append(" not in (");
+ } else {
+ sql.append(" in (");
+ }
+
+ for (String conditionValue : conditionValues) {
+ if (serdeConstants.STRING_TYPE_NAME.equals(typeName)) {
+ sql.append("'").append(conditionValue).append("'");
+ } else if (serdeConstants.DATE_TYPE_NAME.equals(typeName)) {
+ sql.append("to_date('").append(conditionValue).append("')");
+ } else if (serdeConstants.TIMESTAMP_TYPE_NAME.equals(typeName)) {
+ sql.append("to_timestamp('").append(conditionValue).append("')");
+ } else {
+ sql.append(conditionValue);
+ }
+
+ sql.append(", ");
+ }
+
+ sql.delete(sql.length() - 2, sql.length());
+
+ sql.append(")");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/537b90be/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixConnectionUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixConnectionUtil.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixConnectionUtil.java
new file mode 100644
index 0000000..51f6c7e
--- /dev/null
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixConnectionUtil.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.phoenix.hive.util;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.phoenix.hive.constants.PhoenixStorageHandlerConstants;
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.QueryUtil;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Set of methods to obtain Connection depending on configuration
+ */
+
+public class PhoenixConnectionUtil {
+
+ private static final Log LOG = LogFactory.getLog(PhoenixConnectionUtil.class);
+
+ public static Connection getInputConnection(final Configuration conf, final Properties props)
+ throws SQLException {
+ String quorum = conf.get(PhoenixStorageHandlerConstants.ZOOKEEPER_QUORUM);
+ quorum = quorum == null ? props.getProperty(PhoenixStorageHandlerConstants
+ .ZOOKEEPER_QUORUM, PhoenixStorageHandlerConstants.DEFAULT_ZOOKEEPER_QUORUM) :
+ quorum;
+
+ int zooKeeperClientPort = conf.getInt(PhoenixStorageHandlerConstants.ZOOKEEPER_PORT, 0);
+ zooKeeperClientPort = zooKeeperClientPort == 0 ?
+ Integer.parseInt(props.getProperty(PhoenixStorageHandlerConstants.ZOOKEEPER_PORT,
+ String.valueOf(PhoenixStorageHandlerConstants.DEFAULT_ZOOKEEPER_PORT))) :
+ zooKeeperClientPort;
+
+ String zNodeParent = conf.get(PhoenixStorageHandlerConstants.ZOOKEEPER_PARENT);
+ zNodeParent = zNodeParent == null ? props.getProperty(PhoenixStorageHandlerConstants
+ .ZOOKEEPER_PARENT, PhoenixStorageHandlerConstants.DEFAULT_ZOOKEEPER_PARENT) :
+ zNodeParent;
+
+ return getConnection(quorum, zooKeeperClientPort, zNodeParent, PropertiesUtil
+ .extractProperties(props, conf));
+ }
+
+ public static Connection getConnection(final Table table) throws SQLException {
+ Map<String, String> tableParameterMap = table.getParameters();
+
+ String zookeeperQuorum = tableParameterMap.get(PhoenixStorageHandlerConstants
+ .ZOOKEEPER_QUORUM);
+ zookeeperQuorum = zookeeperQuorum == null ? PhoenixStorageHandlerConstants
+ .DEFAULT_ZOOKEEPER_QUORUM : zookeeperQuorum;
+
+ String clientPortString = tableParameterMap.get(PhoenixStorageHandlerConstants
+ .ZOOKEEPER_PORT);
+ int clientPort = clientPortString == null ? PhoenixStorageHandlerConstants
+ .DEFAULT_ZOOKEEPER_PORT : Integer.parseInt(clientPortString);
+
+ String zNodeParent = tableParameterMap.get(PhoenixStorageHandlerConstants.ZOOKEEPER_PARENT);
+ zNodeParent = zNodeParent == null ? PhoenixStorageHandlerConstants
+ .DEFAULT_ZOOKEEPER_PARENT : zNodeParent;
+
+ return DriverManager.getConnection(QueryUtil.getUrl(zookeeperQuorum, clientPort,
+ zNodeParent));
+ }
+
+ private static Connection getConnection(final String quorum, final Integer clientPort, String
+ zNodeParent, Properties props) throws SQLException {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Connection attrs [quorum, port, znode] : " + quorum + ", " + clientPort +
+ ", " +
+ zNodeParent);
+ }
+
+ return DriverManager.getConnection(clientPort != null ? QueryUtil.getUrl(quorum,
+ clientPort, zNodeParent) : QueryUtil.getUrl(quorum), props);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/537b90be/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixStorageHandlerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixStorageHandlerUtil.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixStorageHandlerUtil.java
new file mode 100644
index 0000000..eb5fd24
--- /dev/null
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/util/PhoenixStorageHandlerUtil.java
@@ -0,0 +1,278 @@
+/*
+ * 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.phoenix.hive.util;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.Maps;
+import org.apache.commons.logging.Log;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HRegionLocation;
+import org.apache.hadoop.hbase.util.Strings;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.hive.ql.io.AcidOutputFormat.Options;
+import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.net.DNS;
+import org.apache.phoenix.hive.constants.PhoenixStorageHandlerConstants;
+import org.apache.phoenix.hive.ql.index.IndexSearchCondition;
+
+import javax.naming.NamingException;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
+
+/**
+ * Misc utils for PhoenixStorageHandler
+ */
+
+public class PhoenixStorageHandlerUtil {
+
+ public static String getTargetTableName(Table table) {
+ Map<String, String> tableParameterMap = table.getParameters();
+ String tableName = tableParameterMap.get(PhoenixStorageHandlerConstants
+ .PHOENIX_TABLE_NAME);
+ if (tableName == null) {
+ tableName = table.getTableName();
+ tableParameterMap.put(PhoenixStorageHandlerConstants.PHOENIX_TABLE_NAME, tableName);
+ }
+
+ return tableName;
+ }
+
+
+ public static Object[] toTypedValues(JobConf jobConf, String typeName, String[] values) throws
+ Exception {
+ Object[] results = new Object[values.length];
+ DateFormat df = null;
+
+ for (int i = 0, limit = values.length; i < limit; i++) {
+ if (serdeConstants.STRING_TYPE_NAME.equals(typeName)) {
+ results[i] = values[i];
+ } else if (serdeConstants.INT_TYPE_NAME.equals(typeName)) {
+ results[i] = new Integer(values[i]);
+ } else if (serdeConstants.BIGINT_TYPE_NAME.equals(typeName)) {
+ results[i] = new Long(values[i]);
+ } else if (serdeConstants.DOUBLE_TYPE_NAME.equals(typeName)) {
+ results[i] = new Double(values[i]);
+ } else if (serdeConstants.FLOAT_TYPE_NAME.equals(typeName)) {
+ results[i] = new Float(values[i]);
+ } else if (serdeConstants.SMALLINT_TYPE_NAME.equals(typeName)) {
+ results[i] = new Short(values[i]);
+ } else if (serdeConstants.TINYINT_TYPE_NAME.equals(typeName)) {
+ results[i] = new Byte(values[i]);
+ } else if (serdeConstants.DATE_TYPE_NAME.equals(typeName)) {
+ String dateFormat = jobConf.get(PhoenixStorageHandlerConstants.HBASE_DATE_FORMAT,
+ PhoenixStorageHandlerConstants.DEFAULT_DATE_FORMAT);
+ df = new SimpleDateFormat(dateFormat);
+ results[i] = new Long(df.parse(values[i]).getTime());
+ } else if (serdeConstants.TIMESTAMP_TYPE_NAME.equals(typeName)) {
+ String timestampFormat = jobConf.get(PhoenixStorageHandlerConstants
+ .HBASE_TIMESTAMP_FORMAT, PhoenixStorageHandlerConstants
+ .DEFAULT_TIMESTAMP_FORMAT);
+ df = new SimpleDateFormat(timestampFormat);
+ results[i] = new Long(df.parse(values[i]).getTime());
+ } else if (typeName.contains(serdeConstants.DECIMAL_TYPE_NAME)) {
+ results[i] = new BigDecimal(values[i]);
+ }
+ }
+
+ return results;
+ }
+
+ public static String[] getConstantValues(IndexSearchCondition condition, String comparisonOp) {
+ String[] constantValues = null;
+
+ if (comparisonOp.endsWith("UDFOPEqual") || comparisonOp.endsWith("UDFOPNotEqual")) {
+ constantValues = new String[]{String.valueOf(condition.getConstantDesc().getValue())};
+ } else if (comparisonOp.endsWith("UDFOPEqualOrGreaterThan")) { // key >= 1
+ constantValues = new String[]{String.valueOf(condition.getConstantDesc().getValue())};
+ } else if (comparisonOp.endsWith("UDFOPGreaterThan")) { // key > 1
+ constantValues = new String[]{String.valueOf(condition.getConstantDesc().getValue())};
+ } else if (comparisonOp.endsWith("UDFOPEqualOrLessThan")) { // key <= 1
+ constantValues = new String[]{String.valueOf(condition.getConstantDesc().getValue())};
+ } else if (comparisonOp.endsWith("UDFOPLessThan")) { // key < 1
+ constantValues = new String[]{String.valueOf(condition.getConstantDesc().getValue())};
+ } else if (comparisonOp.endsWith("GenericUDFBetween")) {
+ constantValues = new String[]{String.valueOf(condition.getConstantDesc(0).getValue()),
+ String.valueOf(condition.getConstantDesc(1).getValue())};
+ } else if (comparisonOp.endsWith("GenericUDFIn")) {
+ ExprNodeConstantDesc[] constantDescs = condition.getConstantDescs();
+ constantValues = new String[constantDescs.length];
+ for (int i = 0, limit = constantDescs.length; i < limit; i++) {
+ constantValues[i] = String.valueOf(condition.getConstantDesc(i).getValue());
+ }
+ }
+
+ return constantValues;
+ }
+
+ public static String getRegionLocation(HRegionLocation location, Log log) throws IOException {
+ InetSocketAddress isa = new InetSocketAddress(location.getHostname(), location.getPort());
+ if (isa.isUnresolved()) {
+ log.warn("Failed resolve " + isa);
+ }
+ InetAddress regionAddress = isa.getAddress();
+ String regionLocation = null;
+ try {
+ regionLocation = reverseDNS(regionAddress);
+ } catch (NamingException e) {
+ log.warn("Cannot resolve the host name for " + regionAddress + " because of " + e);
+ regionLocation = location.getHostname();
+ }
+
+ return regionLocation;
+ }
+
+ // Copy from org.apache.hadoop.hbase.mapreduce.TableInputFormatBase.reverseDNS
+ private static final Map<InetAddress, String> reverseDNSCacheMap = Maps.newConcurrentMap();
+
+ private static String reverseDNS(InetAddress ipAddress) throws NamingException,
+ UnknownHostException {
+ String hostName = reverseDNSCacheMap.get(ipAddress);
+
+ if (hostName == null) {
+ String ipAddressString = null;
+ try {
+ ipAddressString = DNS.reverseDns(ipAddress, null);
+ } catch (Exception e) {
+ // We can use InetAddress in case the jndi failed to pull up the reverse DNS entry
+ // from the name service. Also, in case of ipv6, we need to use the InetAddress
+ // since resolving reverse DNS using jndi doesn't work well with ipv6 addresses.
+ ipAddressString = InetAddress.getByName(ipAddress.getHostAddress()).getHostName();
+ }
+
+ if (ipAddressString == null) {
+ throw new UnknownHostException("No host found for " + ipAddress);
+ }
+
+ hostName = Strings.domainNamePointerToHostName(ipAddressString);
+ reverseDNSCacheMap.put(ipAddress, hostName);
+ }
+
+ return hostName;
+ }
+
+ public static String getTableKeyOfSession(JobConf jobConf, String tableName) {
+ SessionState sessionState = SessionState.get();
+
+ String sessionId = sessionState.getSessionId();
+
+ return new StringBuilder("[").append(sessionId).append("]-").append(tableName).toString();
+ }
+
+ public static Map<String, String> createColumnTypeMap(JobConf jobConf) {
+ Map<String, String> columnTypeMap = Maps.newHashMap();
+
+ String[] columnNames = jobConf.get(serdeConstants.LIST_COLUMNS).split
+ (PhoenixStorageHandlerConstants.COMMA);
+ String[] columnTypes = jobConf.get(serdeConstants.LIST_COLUMN_TYPES).split
+ (PhoenixStorageHandlerConstants.COMMA);
+
+ for (int i = 0, limit = columnNames.length; i < limit; i++) {
+ columnTypeMap.put(columnNames[i], columnTypes[i]);
+ }
+
+ return columnTypeMap;
+ }
+
+ public static boolean isTransactionalTable(Properties tableProperties) {
+ String tableIsTransactional = tableProperties.getProperty(hive_metastoreConstants
+ .TABLE_IS_TRANSACTIONAL);
+
+ return tableIsTransactional != null && tableIsTransactional.equalsIgnoreCase("true");
+ }
+
+ public static boolean isTransactionalTable(Configuration config) {
+ String tableIsTransactional = config.get(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL);
+
+ return tableIsTransactional != null && tableIsTransactional.equalsIgnoreCase("true");
+ }
+
+ public static void printConfiguration(Configuration config) {
+ if (Boolean.getBoolean("dev")) {
+ for (Iterator<Entry<String, String>> iterator = config.iterator(); iterator.hasNext();
+ ) {
+ Entry<String, String> entry = iterator.next();
+
+ System.out.println(entry.getKey() + "=" + entry.getValue());
+ }
+ }
+ }
+
+ public static String toString(Object obj) {
+ String content = null;
+
+ if (obj instanceof Array) {
+ Object[] values = (Object[]) obj;
+
+ content = Joiner.on(PhoenixStorageHandlerConstants.COMMA).join(values);
+ } else {
+ content = obj.toString();
+ }
+
+ return content;
+ }
+
+ public static Map<?, ?> toMap(byte[] serialized) {
+ Map<?, ?> resultMap = null;
+ ByteArrayInputStream bais = new ByteArrayInputStream(serialized);
+
+ try (ObjectInputStream ois = new ObjectInputStream(bais)) {
+ resultMap = (Map<?, ?>) ois.readObject();
+ } catch (ClassNotFoundException | IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ return resultMap;
+ }
+
+ public static String getOptionsValue(Options options) {
+ StringBuilder content = new StringBuilder();
+
+ int bucket = options.getBucket();
+ String inspectorInfo = options.getInspector().getCategory() + ":" + options.getInspector()
+ .getTypeName();
+ long maxTxnId = options.getMaximumTransactionId();
+ long minTxnId = options.getMinimumTransactionId();
+ int recordIdColumn = options.getRecordIdColumn();
+ boolean isCompresses = options.isCompressed();
+ boolean isWritingBase = options.isWritingBase();
+
+ content.append("bucket : ").append(bucket).append(", inspectorInfo : ").append
+ (inspectorInfo).append(", minTxnId : ").append(minTxnId).append(", maxTxnId : ")
+ .append(maxTxnId).append(", recordIdColumn : ").append(recordIdColumn);
+ content.append(", isCompressed : ").append(isCompresses).append(", isWritingBase : ")
+ .append(isWritingBase);
+
+ return content.toString();
+ }
+}