Repository: tajo Updated Branches: refs/heads/branch-0.11.0 77ac53c37 -> 13569f005
http://git-wip-us.apache.org/repos/asf/tajo/blob/13569f00/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java index fc0b1bb..79cdaa3 100644 --- a/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java +++ b/tajo-plan/src/main/java/org/apache/tajo/plan/rewrite/rules/PartitionedTableRewriter.java @@ -18,25 +18,26 @@ package org.apache.tajo.plan.rewrite.rules; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Lists; import com.google.common.collect.Sets; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.hadoop.fs.*; import org.apache.tajo.OverridableConf; -import org.apache.tajo.catalog.*; +import org.apache.tajo.catalog.Column; +import org.apache.tajo.catalog.Schema; +import org.apache.tajo.catalog.TableDesc; import org.apache.tajo.catalog.partition.PartitionMethodDesc; -import org.apache.tajo.catalog.proto.CatalogProtos.PartitionsByAlgebraProto; -import org.apache.tajo.catalog.proto.CatalogProtos.PartitionDescProto; import org.apache.tajo.datum.DatumFactory; import org.apache.tajo.datum.NullDatum; -import org.apache.tajo.exception.*; +import org.apache.tajo.exception.TajoException; +import org.apache.tajo.exception.TajoInternalError; import org.apache.tajo.plan.LogicalPlan; import org.apache.tajo.plan.expr.*; import org.apache.tajo.plan.logical.*; import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRule; import org.apache.tajo.plan.rewrite.LogicalPlanRewriteRuleContext; -import org.apache.tajo.plan.util.EvalNodeToExprConverter; import org.apache.tajo.plan.util.PlannerUtil; import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor; import org.apache.tajo.storage.Tuple; @@ -44,12 +45,11 @@ import org.apache.tajo.storage.VTuple; import org.apache.tajo.util.StringUtils; import java.io.IOException; -import java.util.*; +import java.util.List; +import java.util.Set; +import java.util.Stack; public class PartitionedTableRewriter implements LogicalPlanRewriteRule { - private CatalogService catalog; - private long totalVolume; - private static final Log LOG = LogFactory.getLog(PartitionedTableRewriter.class); private static final String NAME = "Partitioned Table Rewriter"; @@ -79,7 +79,6 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { public LogicalPlan rewrite(LogicalPlanRewriteRuleContext context) throws TajoException { LogicalPlan plan = context.getPlan(); LogicalPlan.QueryBlock rootBlock = plan.getRootBlock(); - this.catalog = context.getCatalog(); rewriter.visit(context.getQueryContext(), plan, rootBlock, rootBlock.getRoot(), new Stack<LogicalNode>()); return plan; } @@ -110,13 +109,6 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { } } - private Path [] findFilteredPaths(OverridableConf queryContext, String tableName, - Schema partitionColumns, EvalNode [] conjunctiveForms, Path tablePath) - throws IOException, UndefinedDatabaseException, UndefinedTableException, UndefinedPartitionMethodException, - UndefinedOperatorException, UnsupportedException { - return findFilteredPaths(queryContext, tableName, partitionColumns, conjunctiveForms, tablePath, null); - } - /** * It assumes that each conjunctive form corresponds to one column. * @@ -127,82 +119,13 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { * @return * @throws IOException */ - private Path [] findFilteredPaths(OverridableConf queryContext, String tableName, - Schema partitionColumns, EvalNode [] conjunctiveForms, Path tablePath, ScanNode scanNode) - throws IOException, UndefinedDatabaseException, UndefinedTableException, UndefinedPartitionMethodException, - UndefinedOperatorException, UnsupportedException { + private Path [] findFilteredPaths(OverridableConf queryContext, Schema partitionColumns, EvalNode [] conjunctiveForms, + Path tablePath) + throws IOException { - Path [] filteredPaths = null; FileSystem fs = tablePath.getFileSystem(queryContext.getConf()); - String [] splits = CatalogUtil.splitFQTableName(tableName); - List<PartitionDescProto> partitions = null; - - try { - if (conjunctiveForms == null) { - partitions = catalog.getPartitionsOfTable(splits[0], splits[1]); - if (partitions.isEmpty()) { - filteredPaths = findFilteredPathsFromFileSystem(partitionColumns, conjunctiveForms, fs, tablePath); - } else { - filteredPaths = findFilteredPathsByPartitionDesc(partitions); - } - } else { - if (catalog.existPartitions(splits[0], splits[1])) { - PartitionsByAlgebraProto request = getPartitionsAlgebraProto(splits[0], splits[1], conjunctiveForms); - partitions = catalog.getPartitionsByAlgebra(request); - filteredPaths = findFilteredPathsByPartitionDesc(partitions); - } else { - filteredPaths = findFilteredPathsFromFileSystem(partitionColumns, conjunctiveForms, fs, tablePath); - } - } - } catch (UnsupportedException ue) { - // Partial catalog might not allow some filter conditions. For example, HiveMetastore doesn't In statement, - // regexp statement and so on. Above case, Tajo need to build filtered path by listing hdfs directories. - LOG.warn(ue.getMessage()); - partitions = catalog.getPartitionsOfTable(splits[0], splits[1]); - if (partitions.isEmpty()) { - filteredPaths = findFilteredPathsFromFileSystem(partitionColumns, conjunctiveForms, fs, tablePath); - } else { - filteredPaths = findFilteredPathsByPartitionDesc(partitions); - } - scanNode.setQual(AlgebraicUtil.createSingletonExprFromCNF(conjunctiveForms)); - } - LOG.info("Filtered directory or files: " + filteredPaths.length); - return filteredPaths; - } - - /** - * Build list of partition path by PartitionDescProto which is generated from CatalogStore. - * - * @param partitions - * @return - */ - private Path[] findFilteredPathsByPartitionDesc(List<PartitionDescProto> partitions) { - Path [] filteredPaths = new Path[partitions.size()]; - for (int i = 0; i < partitions.size(); i++) { - PartitionDescProto partition = partitions.get(i); - filteredPaths[i] = new Path(partition.getPath()); - totalVolume += partition.getNumBytes(); - } - return filteredPaths; - } - - /** - * Build list of partition path by filtering directories in the given table path. - * - * - * @param partitionColumns - * @param conjunctiveForms - * @param fs - * @param tablePath - * @return - * @throws IOException - */ - private Path [] findFilteredPathsFromFileSystem(Schema partitionColumns, EvalNode [] conjunctiveForms, - FileSystem fs, Path tablePath) throws IOException{ - Path [] filteredPaths = null; PathFilter [] filters; - if (conjunctiveForms == null) { filters = buildAllAcceptingPathFilters(partitionColumns); } else { @@ -210,40 +133,15 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { } // loop from one to the number of partition columns - filteredPaths = toPathArray(fs.listStatus(tablePath, filters[0])); + Path [] filteredPaths = toPathArray(fs.listStatus(tablePath, filters[0])); for (int i = 1; i < partitionColumns.size(); i++) { // Get all file status matched to a ith level path filter. filteredPaths = toPathArray(fs.listStatus(filteredPaths, filters[i])); } - return filteredPaths; - } - - /** - * Build algebra expressions for querying partitions and partition keys by using EvalNodeToExprConverter. - * - * @param databaseName the database name - * @param tableName the table name - * @param conjunctiveForms EvalNode which contains filter conditions - * @return - */ - public static PartitionsByAlgebraProto getPartitionsAlgebraProto( - String databaseName, String tableName, EvalNode [] conjunctiveForms) { - - PartitionsByAlgebraProto.Builder request = PartitionsByAlgebraProto.newBuilder(); - request.setDatabaseName(databaseName); - request.setTableName(tableName); - - if (conjunctiveForms != null) { - EvalNode evalNode = AlgebraicUtil.createSingletonExprFromCNF(conjunctiveForms); - EvalNodeToExprConverter convertor = new EvalNodeToExprConverter(databaseName + "." + tableName); - convertor.visit(null, evalNode, new Stack<EvalNode>()); - request.setAlgebra(convertor.getResult().toJson()); - } else { - request.setAlgebra(""); - } - return request.build(); + LOG.info("Filtered directory or files: " + filteredPaths.length); + return filteredPaths; } /** @@ -294,7 +192,6 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { accumulatedFilters.toArray(new EvalNode[accumulatedFilters.size()])); filters[i] = new PartitionPathFilter(partitionColumns, filterPerLevel); } - return filters; } @@ -318,19 +215,16 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { return filters; } - private Path [] toPathArray(FileStatus[] fileStatuses) { + public static Path [] toPathArray(FileStatus[] fileStatuses) { Path [] paths = new Path[fileStatuses.length]; - for (int i = 0; i < fileStatuses.length; i++) { - FileStatus fileStatus = fileStatuses[i]; - paths[i] = fileStatus.getPath(); - totalVolume += fileStatus.getLen(); + for (int j = 0; j < fileStatuses.length; j++) { + paths[j] = fileStatuses[j].getPath(); } return paths; } - public Path [] findFilteredPartitionPaths(OverridableConf queryContext, ScanNode scanNode) throws IOException, - UndefinedDatabaseException, UndefinedTableException, UndefinedPartitionMethodException, - UndefinedOperatorException, UnsupportedException { + @VisibleForTesting + public Path [] findFilteredPartitionPaths(OverridableConf queryContext, ScanNode scanNode) throws IOException { TableDesc table = scanNode.getTableDesc(); PartitionMethodDesc partitionDesc = scanNode.getTableDesc().getPartitionMethod(); @@ -369,10 +263,10 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { } if (indexablePredicateSet.size() > 0) { // There are at least one indexable predicates - return findFilteredPaths(queryContext, table.getName(), paritionValuesSchema, - indexablePredicateSet.toArray(new EvalNode[indexablePredicateSet.size()]), new Path(table.getUri()), scanNode); + return findFilteredPaths(queryContext, paritionValuesSchema, + indexablePredicateSet.toArray(new EvalNode[indexablePredicateSet.size()]), new Path(table.getUri())); } else { // otherwise, we will get all partition paths. - return findFilteredPaths(queryContext, table.getName(), paritionValuesSchema, null, new Path(table.getUri())); + return findFilteredPaths(queryContext, paritionValuesSchema, null, new Path(table.getUri())); } } @@ -421,6 +315,24 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { } } + private void updateTableStat(OverridableConf queryContext, PartitionedTableScanNode scanNode) + throws TajoException { + if (scanNode.getInputPaths().length > 0) { + try { + FileSystem fs = scanNode.getInputPaths()[0].getFileSystem(queryContext.getConf()); + long totalVolume = 0; + + for (Path input : scanNode.getInputPaths()) { + ContentSummary summary = fs.getContentSummary(input); + totalVolume += summary.getLength(); + } + scanNode.getTableDesc().getStats().setNumBytes(totalVolume); + } catch (Throwable e) { + throw new TajoInternalError(e); + } + } + } + /** * Take a look at a column partition path. A partition path consists * of a table path part and column values part. This method transforms @@ -500,7 +412,7 @@ public class PartitionedTableRewriter implements LogicalPlanRewriteRule { plan.addHistory("PartitionTableRewriter chooses " + filteredPaths.length + " of partitions"); PartitionedTableScanNode rewrittenScanNode = plan.createNode(PartitionedTableScanNode.class); rewrittenScanNode.init(scanNode, filteredPaths); - rewrittenScanNode.getTableDesc().getStats().setNumBytes(totalVolume); + updateTableStat(queryContext, rewrittenScanNode); // if it is topmost node, set it as the rootnode of this block. if (stack.empty() || block.getRoot().equals(scanNode)) { http://git-wip-us.apache.org/repos/asf/tajo/blob/13569f00/tajo-plan/src/main/java/org/apache/tajo/plan/util/EvalNodeToExprConverter.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/util/EvalNodeToExprConverter.java b/tajo-plan/src/main/java/org/apache/tajo/plan/util/EvalNodeToExprConverter.java deleted file mode 100644 index c656671..0000000 --- a/tajo-plan/src/main/java/org/apache/tajo/plan/util/EvalNodeToExprConverter.java +++ /dev/null @@ -1,297 +0,0 @@ -/** - * 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.tajo.plan.util; - -import org.apache.tajo.algebra.*; -import org.apache.tajo.datum.DateDatum; -import org.apache.tajo.datum.Datum; -import org.apache.tajo.datum.TimeDatum; -import org.apache.tajo.datum.TimestampDatum; -import org.apache.tajo.plan.expr.*; - -import java.util.Stack; - -/** - * This converts EvalNode tree to Expr tree. - * - */ -public class EvalNodeToExprConverter extends SimpleEvalNodeVisitor<Object> { - private Stack<Expr> exprs = new Stack<Expr>(); - - private String tableName; - - public EvalNodeToExprConverter(String tableName) { - this.tableName = tableName; - } - - public Expr getResult() { - return exprs.pop(); - } - - @Override - protected EvalNode visitBinaryEval(Object o, Stack<EvalNode> stack, BinaryEval binaryEval) { - stack.push(binaryEval); - visit(o, binaryEval.getLeftExpr(), stack); - Expr left = exprs.pop(); - - visit(o, binaryEval.getRightExpr(), stack); - Expr right = exprs.pop(); - - Expr expr = null; - switch (binaryEval.getType()) { - // Arithmetic expression - case PLUS: - expr = new BinaryOperator(OpType.Plus, left, right); - break; - case MINUS: - expr = new BinaryOperator(OpType.Minus, left, right); - break; - case MULTIPLY: - expr = new BinaryOperator(OpType.Multiply, left, right); - break; - case DIVIDE: - expr = new BinaryOperator(OpType.Divide, left, right); - break; - case MODULAR: - expr = new BinaryOperator(OpType.Modular, left, right); - break; - - // Logical Predicates - case AND: - expr = new BinaryOperator(OpType.And, left, right); - break; - case OR: - expr = new BinaryOperator(OpType.Or, left, right); - break; - case NOT: - expr = new BinaryOperator(OpType.Not, left, right); - break; - - // Comparison Predicates - case EQUAL: - expr = new BinaryOperator(OpType.Equals, left, right); - break; - case NOT_EQUAL: - expr = new BinaryOperator(OpType.NotEquals, left, right); - break; - case LTH: - expr = new BinaryOperator(OpType.LessThan, left, right); - break; - case LEQ: - expr = new BinaryOperator(OpType.LessThanOrEquals, left, right); - break; - case GTH: - expr = new BinaryOperator(OpType.GreaterThan, left, right); - break; - case GEQ: - expr = new BinaryOperator(OpType.GreaterThanOrEquals, left, right); - break; - - // SQL standard predicates - case IS_NULL: - expr = new BinaryOperator(OpType.IsNullPredicate, left, right); - break; - case CASE: - expr = new BinaryOperator(OpType.CaseWhen, left, right); - break; - case IN: - InEval inEval = (InEval) binaryEval; - expr = new InPredicate(left, right, inEval.isNot()); - break; - - // String operators and Pattern match predicates - case LIKE: - LikePredicateEval likePredicateEval = (LikePredicateEval) binaryEval; - expr = new PatternMatchPredicate(OpType.LikePredicate, likePredicateEval.isNot(), left, right); - break; - case SIMILAR_TO: - SimilarToPredicateEval similarToPredicateEval = (SimilarToPredicateEval) binaryEval; - expr = new PatternMatchPredicate(OpType.SimilarToPredicate, similarToPredicateEval.isNot(), left, right); - break; - case REGEX: - RegexPredicateEval regexPredicateEval = (RegexPredicateEval) binaryEval; - expr = new PatternMatchPredicate(OpType.Regexp, regexPredicateEval.isNot(), left, right); - break; - case CONCATENATE: - default: - throw new RuntimeException("Unsupported type: " + binaryEval.getType().name()); - } - - if (expr != null) { - exprs.push(expr); - } - - stack.pop(); - return null; - } - - @Override - protected EvalNode visitConst(Object o, ConstEval evalNode, Stack<EvalNode> stack) { - Expr value = null; - DateValue dateValue; - TimeValue timeValue; - - switch (evalNode.getValueType().getType()) { - case NULL_TYPE: - value = new NullLiteral(); - break; - case BOOLEAN: - value = new LiteralValue(evalNode.getValue().asChars(), LiteralValue.LiteralType.Boolean); - break; - case INT1: - case INT2: - case INT4: - value = new LiteralValue(evalNode.getValue().asChars(), LiteralValue.LiteralType.Unsigned_Integer); - break; - case INT8: - value = new LiteralValue(evalNode.getValue().asChars(), LiteralValue.LiteralType.Unsigned_Large_Integer); - break; - case FLOAT4: - case FLOAT8: - value = new LiteralValue(evalNode.getValue().asChars(), LiteralValue.LiteralType.Unsigned_Float); - break; - case TEXT: - value = new LiteralValue(evalNode.getValue().asChars(), LiteralValue.LiteralType.String); - break; - case DATE: - DateDatum dateDatum = (DateDatum) evalNode.getValue(); - - dateValue = new DateValue(""+dateDatum.getYear(), - ""+dateDatum.getMonthOfYear(), ""+dateDatum.getDayOfMonth()); - value = new DateLiteral(dateValue); - - break; - case TIMESTAMP: - TimestampDatum timestampDatum = (TimestampDatum) evalNode.getValue(); - - dateValue = new DateValue(""+timestampDatum.getYear(), - ""+timestampDatum.getMonthOfYear(), ""+timestampDatum.getDayOfMonth()); - - timeValue = new TimeValue(""+timestampDatum.getHourOfDay() - , ""+timestampDatum.getMinuteOfHour(), ""+timestampDatum.getSecondOfMinute()); - - value = new TimestampLiteral(dateValue, timeValue); - break; - case TIME: - TimeDatum timeDatum = (TimeDatum) evalNode.getValue(); - timeValue = new TimeValue(""+timeDatum.getHourOfDay() - , ""+timeDatum.getMinuteOfHour(), ""+timeDatum.getSecondOfMinute()); - - value = new TimeLiteral(timeValue); - break; - default: - throw new RuntimeException("Unsupported type: " + evalNode.getValueType().getType().name()); - } - exprs.push(value); - - return super.visitConst(o, evalNode, stack); - } - - @Override - protected EvalNode visitRowConstant(Object o, RowConstantEval evalNode, Stack<EvalNode> stack) { - Expr[] values = new Expr[evalNode.getValues().length]; - for (int i = 0; i < evalNode.getValues().length; i++) { - Datum datum = evalNode.getValues()[i]; - LiteralValue value; - switch (datum.type()) { - case BOOLEAN: - value = new LiteralValue(datum.asChars(), LiteralValue.LiteralType.Boolean); - break; - case TEXT: - value = new LiteralValue(datum.asChars(), LiteralValue.LiteralType.String); - break; - case INT1: - case INT2: - case INT4: - value = new LiteralValue(datum.asChars(), LiteralValue.LiteralType.Unsigned_Integer); - break; - case INT8: - value = new LiteralValue(datum.asChars(), LiteralValue.LiteralType.Unsigned_Large_Integer); - break; - case FLOAT4: - case FLOAT8: - value = new LiteralValue(datum.asChars(), LiteralValue.LiteralType.Unsigned_Float); - break; - default: - throw new RuntimeException("Unsupported type: " + datum.type().name()); - } - values[i] = value; - } - ValueListExpr expr = new ValueListExpr(values); - exprs.push(expr); - - return super.visitRowConstant(o, evalNode, stack); - } - - @Override - protected EvalNode visitField(Object o, FieldEval evalNode, Stack<EvalNode> stack) { - ColumnReferenceExpr expr = new ColumnReferenceExpr(tableName, evalNode.getColumnName()); - exprs.push(expr); - return super.visitField(o, evalNode, stack); - } - - @Override - protected EvalNode visitBetween(Object o, BetweenPredicateEval evalNode, Stack<EvalNode> stack) { - stack.push(evalNode); - - visit(o, evalNode.getPredicand(), stack); - Expr predicand = exprs.pop(); - - visit(o, evalNode.getBegin(), stack); - Expr begin = exprs.pop(); - - visit(o, evalNode.getEnd(), stack); - Expr end = exprs.pop(); - - Expr expr = new BetweenPredicate(evalNode.isNot(), evalNode.isSymmetric(), predicand, begin, end); - exprs.push(expr); - - stack.pop(); - - return null; - } - - @Override - protected EvalNode visitCaseWhen(Object o, CaseWhenEval evalNode, Stack<EvalNode> stack) { - stack.push(evalNode); - - CaseWhenPredicate caseWhenPredicate = new CaseWhenPredicate(); - - for (CaseWhenEval.IfThenEval ifThenEval : evalNode.getIfThenEvals()) { - visit(o, ifThenEval.getCondition(), stack); - Expr condition = exprs.pop(); - visit(o, ifThenEval.getResult(), stack); - Expr result = exprs.pop(); - - caseWhenPredicate.addWhen(condition, result); - } - - if (evalNode.hasElse()) { - visit(o, evalNode.getElse(), stack); - Expr elseResult = exprs.pop(); - caseWhenPredicate.setElseResult(elseResult); - } - - exprs.push(caseWhenPredicate); - - stack.pop(); - - return null; - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/tajo/blob/13569f00/tajo-plan/src/main/java/org/apache/tajo/plan/util/PartitionFilterAlgebraVisitor.java ---------------------------------------------------------------------- diff --git a/tajo-plan/src/main/java/org/apache/tajo/plan/util/PartitionFilterAlgebraVisitor.java b/tajo-plan/src/main/java/org/apache/tajo/plan/util/PartitionFilterAlgebraVisitor.java deleted file mode 100644 index 72fd939..0000000 --- a/tajo-plan/src/main/java/org/apache/tajo/plan/util/PartitionFilterAlgebraVisitor.java +++ /dev/null @@ -1,573 +0,0 @@ -/** - * 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.tajo.plan.util; - -import org.apache.tajo.algebra.*; -import org.apache.tajo.catalog.CatalogConstants; -import org.apache.tajo.catalog.Column; -import org.apache.tajo.common.TajoDataTypes.*; -import org.apache.tajo.datum.TimeDatum; -import org.apache.tajo.exception.TajoException; -import org.apache.tajo.exception.UnsupportedException; -import org.apache.tajo.plan.ExprAnnotator; -import org.apache.tajo.plan.visitor.SimpleAlgebraVisitor; -import org.apache.tajo.util.Pair; -import org.apache.tajo.util.TUtil; -import org.apache.tajo.util.datetime.DateTimeUtil; -import org.apache.tajo.util.datetime.TimeMeta; - -import java.sql.Date; -import java.sql.Time; -import java.sql.Timestamp; -import java.util.List; -import java.util.Stack; -import java.util.TimeZone; - -/** - * This build SQL statements for getting partitions informs on CatalogStore with algebra expressions. - * This visitor assumes that all columns of algebra expressions are reserved for one table. - * - */ -public class PartitionFilterAlgebraVisitor extends SimpleAlgebraVisitor<Object, Expr> { - private String tableAlias; - private Column column; - private boolean isHiveCatalog = false; - - private Stack<String> queries = new Stack(); - private List<Pair<Type, Object>> parameters = TUtil.newList(); - - public String getTableAlias() { - return tableAlias; - } - - public void setTableAlias(String tableAlias) { - this.tableAlias = tableAlias; - } - - public Column getColumn() { - return column; - } - - public void setColumn(Column column) { - this.column = column; - } - - public boolean isHiveCatalog() { - return isHiveCatalog; - } - - public void setIsHiveCatalog(boolean isHiveCatalog) { - this.isHiveCatalog = isHiveCatalog; - } - - public List<Pair<Type, Object>> getParameters() { - return parameters; - } - - public void setParameters(List<Pair<Type, Object>> parameters) { - this.parameters = parameters; - } - - public void clearParameters() { - this.parameters.clear(); - } - - public String getResult() { - return queries.pop(); - } - - @Override - public Expr visit(Object ctx, Stack<Expr> stack, Expr expr) throws TajoException { - if (expr.getType() == OpType.LikePredicate) { - return visitLikePredicate(ctx, stack, (PatternMatchPredicate) expr); - } else if (expr.getType() == OpType.SimilarToPredicate) { - return visitSimilarToPredicate(ctx, stack, (PatternMatchPredicate) expr); - } else if (expr.getType() == OpType.Regexp) { - return visitRegexpPredicate(ctx, stack, (PatternMatchPredicate) expr); - } - return super.visit(ctx, stack, expr); - } - - @Override - public Expr visitDateLiteral(Object ctx, Stack<Expr> stack, DateLiteral expr) throws TajoException { - StringBuilder sb = new StringBuilder(); - - if (!isHiveCatalog) { - sb.append("?").append(" )"); - parameters.add(new Pair(Type.DATE, Date.valueOf(expr.toString()))); - } else { - sb.append("\"").append(expr.toString()).append("\""); - } - queries.push(sb.toString()); - return expr; - } - - @Override - public Expr visitTimestampLiteral(Object ctx, Stack<Expr> stack, TimestampLiteral expr) throws TajoException { - StringBuilder sb = new StringBuilder(); - - if (!isHiveCatalog) { - DateValue dateValue = expr.getDate(); - TimeValue timeValue = expr.getTime(); - - int [] dates = ExprAnnotator.dateToIntArray(dateValue.getYears(), - dateValue.getMonths(), - dateValue.getDays()); - int [] times = ExprAnnotator.timeToIntArray(timeValue.getHours(), - timeValue.getMinutes(), - timeValue.getSeconds(), - timeValue.getSecondsFraction()); - - long julianTimestamp; - if (timeValue.hasSecondsFraction()) { - julianTimestamp = DateTimeUtil.toJulianTimestamp(dates[0], dates[1], dates[2], times[0], times[1], times[2], - times[3] * 1000); - } else { - julianTimestamp = DateTimeUtil.toJulianTimestamp(dates[0], dates[1], dates[2], times[0], times[1], times[2], 0); - } - - TimeMeta tm = new TimeMeta(); - DateTimeUtil.toJulianTimeMeta(julianTimestamp, tm); - - TimeZone tz = TimeZone.getDefault(); - DateTimeUtil.toUTCTimezone(tm, tz); - - sb.append("?").append(" )"); - Timestamp timestamp = new Timestamp(DateTimeUtil.julianTimeToJavaTime(DateTimeUtil.toJulianTimestamp(tm))); - parameters.add(new Pair(Type.TIMESTAMP, timestamp)); - } else { - sb.append("\"").append(expr.toString()).append("\""); - } - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitTimeLiteral(Object ctx, Stack<Expr> stack, TimeLiteral expr) throws TajoException { - StringBuilder sb = new StringBuilder(); - - if (!isHiveCatalog) { - TimeValue timeValue = expr.getTime(); - - int [] times = ExprAnnotator.timeToIntArray(timeValue.getHours(), - timeValue.getMinutes(), - timeValue.getSeconds(), - timeValue.getSecondsFraction()); - - long time; - if (timeValue.hasSecondsFraction()) { - time = DateTimeUtil.toTime(times[0], times[1], times[2], times[3] * 1000); - } else { - time = DateTimeUtil.toTime(times[0], times[1], times[2], 0); - } - TimeDatum timeDatum = new TimeDatum(time); - TimeMeta tm = timeDatum.asTimeMeta(); - - TimeZone tz = TimeZone.getDefault(); - DateTimeUtil.toUTCTimezone(tm, tz); - - sb.append("?").append(" )"); - parameters.add(new Pair(Type.TIME, new Time(DateTimeUtil.toJavaTime(tm.hours, tm.minutes, tm.secs, tm.fsecs)))); - } else { - sb.append("\"").append(expr.toString()).append("\""); - } - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitLiteral(Object ctx, Stack<Expr> stack, LiteralValue expr) throws TajoException { - StringBuilder sb = new StringBuilder(); - - if (!isHiveCatalog) { - sb.append("?").append(" )"); - switch (expr.getValueType()) { - case Boolean: - parameters.add(new Pair(Type.BOOLEAN, Boolean.valueOf(expr.getValue()))); - break; - case Unsigned_Float: - parameters.add(new Pair(Type.FLOAT8, Double.valueOf(expr.getValue()))); - break; - case String: - parameters.add(new Pair(Type.TEXT, expr.getValue())); - break; - default: - parameters.add(new Pair(Type.INT8, Long.valueOf(expr.getValue()))); - break; - } - } else { - switch (expr.getValueType()) { - case String: - sb.append("\"").append(expr.getValue()).append("\""); - break; - default: - sb.append(expr.getValue()); - break; - } - } - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitValueListExpr(Object ctx, Stack<Expr> stack, ValueListExpr expr) throws TajoException { - StringBuilder sb = new StringBuilder(); - - if (!isHiveCatalog) { - sb.append("("); - for(int i = 0; i < expr.getValues().length; i++) { - if (i > 0) { - sb.append(", "); - } - sb.append("?"); - stack.push(expr.getValues()[i]); - visit(ctx, stack, expr.getValues()[i]); - stack.pop(); - } - sb.append(")"); - sb.append(" )"); - queries.push(sb.toString()); - } else { - throw new UnsupportedException("IN Operator"); - } - - return expr; - } - - @Override - public Expr visitColumnReference(Object ctx, Stack<Expr> stack, ColumnReferenceExpr expr) throws TajoException { - StringBuilder sb = new StringBuilder(); - - if (!isHiveCatalog) { - sb.append("( ").append(tableAlias).append(".").append(CatalogConstants.COL_COLUMN_NAME) - .append(" = '").append(expr.getName()).append("'") - .append(" AND ").append(tableAlias).append(".").append(CatalogConstants.COL_PARTITION_VALUE); - } else { - sb.append(expr.getName()); - } - queries.push(sb.toString()); - return expr; - } - - - @Override - public Expr visitUnaryOperator(Object ctx, Stack<Expr> stack, UnaryOperator expr) throws TajoException { - stack.push(expr); - Expr child = visit(ctx, stack, expr.getChild()); - stack.pop(); - - if (child.getType() == OpType.Literal) { - return new NullLiteral(); - } - - String childSql = queries.pop(); - - StringBuilder sb = new StringBuilder(); - if (expr.getType() == OpType.IsNullPredicate) { - IsNullPredicate isNullPredicate = (IsNullPredicate) expr; - sb.append(childSql); - sb.append(" IS "); - if (isNullPredicate.isNot()) { - sb.append("NOT NULL"); - } else { - sb.append("NULL"); - } - } - - if (!isHiveCatalog) { - sb.append(" )"); - } - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitBetween(Object ctx, Stack<Expr> stack, BetweenPredicate expr) throws TajoException { - stack.push(expr); - - visit(ctx, stack, expr.predicand()); - String predicandSql = queries.pop(); - - visit(ctx, stack, expr.begin()); - String beginSql= queries.pop(); - if (!isHiveCatalog && beginSql.endsWith(")")) { - beginSql = beginSql.substring(0, beginSql.length()-1); - } - - visit(ctx, stack, expr.end()); - String endSql = queries.pop(); - if (!isHiveCatalog && endSql.endsWith(")")) { - endSql = beginSql.substring(0, endSql.length()-1); - } - - StringBuilder sb = new StringBuilder(); - sb.append(predicandSql); - sb.append(" BETWEEN "); - sb.append(beginSql); - sb.append(" AND "); - sb.append(endSql); - - if (!isHiveCatalog) { - sb.append(")"); - } - - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitCaseWhen(Object ctx, Stack<Expr> stack, CaseWhenPredicate expr) throws TajoException { - stack.push(expr); - - StringBuilder sb = new StringBuilder(); - sb.append("CASE "); - - String condition, result; - - for (CaseWhenPredicate.WhenExpr when : expr.getWhens()) { - visit(ctx, stack, when.getCondition()); - condition = queries.pop(); - visit(ctx, stack, when.getResult()); - result = queries.pop(); - - String whenSql = condition + " " + result; - if (!isHiveCatalog && whenSql.endsWith(")")) { - whenSql = whenSql.substring(0, whenSql.length()-1); - } - - sb.append(whenSql).append(" "); - } - - if (expr.hasElseResult()) { - visit(ctx, stack, expr.getElseResult()); - String elseSql = queries.pop(); - if (!isHiveCatalog && elseSql.endsWith(")")) { - elseSql = elseSql.substring(0, elseSql.length()-1); - } - - sb.append("ELSE ").append(elseSql).append(" END"); - } - - if (!isHiveCatalog) { - sb.append(")"); - } - - queries.push(sb.toString()); - - stack.pop(); - return expr; - } - - @Override - public Expr visitBinaryOperator(Object ctx, Stack<Expr> stack, BinaryOperator expr) throws TajoException { - stack.push(expr); - Expr lhs = visit(ctx, stack, expr.getLeft()); - String leftSql = queries.pop(); - Expr rhs = visit(ctx, stack, expr.getRight()); - String rightSql = queries.pop(); - stack.pop(); - - if (!expr.getLeft().equals(lhs)) { - expr.setLeft(lhs); - } - if (!expr.getRight().equals(rhs)) { - expr.setRight(rhs); - } - - if (lhs.getType() == OpType.Literal && rhs.getType() == OpType.Literal) { - return new NullLiteral(); - } - - StringBuilder sb = new StringBuilder(); - sb.append(leftSql); - sb.append(" ").append(getOperator(expr.getType())).append(" "); - sb.append(rightSql); - queries.push(sb.toString()); - - return expr; - } - - private String getOperator(OpType type) { - String operator; - switch (type) { - case Not: - operator = "!"; - break; - case And: - operator = "AND"; - break; - case Or: - operator = "OR"; - break; - case Equals: - operator = "="; - break; - case IsNullPredicate: - operator = "IS NULL"; - break; - case NotEquals: - operator = "<>"; - break; - case LessThan: - operator = "<"; - break; - case LessThanOrEquals: - operator = "<="; - break; - case GreaterThan: - operator = ">"; - break; - case GreaterThanOrEquals: - operator = ">="; - break; - case Plus: - operator = "+"; - break; - case Minus: - operator = "-"; - break; - case Modular: - operator = "%"; - break; - case Multiply: - operator = "*"; - break; - case Divide: - operator = "/"; - break; - case LikePredicate: - operator = "LIKE"; - break; - case SimilarToPredicate: - operator = "([.])"; - break; - case InPredicate: - operator = "IN"; - break; - case Asterisk: - operator = "*"; - break; - //TODO: need to check more types. - default: - operator = type.name(); - break; - } - - return operator; - } - - @Override - public Expr visitLikePredicate(Object ctx, Stack<Expr> stack, PatternMatchPredicate expr) throws TajoException { - stack.push(expr); - - visit(ctx, stack, expr.getPredicand()); - String predicand = queries.pop(); - visit(ctx, stack, expr.getPattern()); - String pattern = queries.pop(); - stack.pop(); - - if(isHiveCatalog) { - if (pattern.startsWith("%") || pattern.endsWith("%")) { - throw new UnsupportedException("LIKE Operator with '%'"); - } - } - StringBuilder sb = new StringBuilder(); - sb.append(predicand); - - if (expr.isNot()) { - sb.append(" NOT "); - } - - if (expr.isCaseInsensitive()) { - sb.append(" ILIKE "); - } else { - sb.append(" LIKE "); - } - - - sb.append(pattern); - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitSimilarToPredicate(Object ctx, Stack<Expr> stack, PatternMatchPredicate expr) throws TajoException { - if (isHiveCatalog) { - throw new UnsupportedException("SIMILAR TO Operator"); - } - - stack.push(expr); - - visit(ctx, stack, expr.getPredicand()); - String predicand = queries.pop(); - visit(ctx, stack, expr.getPattern()); - String pattern = queries.pop(); - stack.pop(); - - StringBuilder sb = new StringBuilder(); - sb.append(predicand); - - if (expr.isNot()) { - sb.append(" NOT "); - } - - sb.append(" SIMILAR TO "); - - sb.append(pattern); - queries.push(sb.toString()); - - return expr; - } - - @Override - public Expr visitRegexpPredicate(Object ctx, Stack<Expr> stack, PatternMatchPredicate expr) throws TajoException { - if (isHiveCatalog) { - throw new UnsupportedException("REGEXP Operator"); - } - - stack.push(expr); - - visit(ctx, stack, expr.getPredicand()); - String predicand = queries.pop(); - visit(ctx, stack, expr.getPattern()); - String pattern = queries.pop(); - stack.pop(); - - StringBuilder sb = new StringBuilder(); - sb.append(predicand); - - if (expr.isNot()) { - sb.append(" NOT "); - } - sb.append(" REGEXP "); - - sb.append(pattern); - queries.push(sb.toString()); - - return expr; - } - -} \ No newline at end of file
