| /** |
| * 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.rewrite.rules; |
| |
| import com.google.common.collect.BiMap; |
| import com.google.common.collect.HashBiMap; |
| import com.google.common.collect.ImmutableSet; |
| import com.google.common.collect.Sets; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.tajo.SessionVars; |
| import org.apache.tajo.algebra.JoinType; |
| import org.apache.tajo.catalog.*; |
| import org.apache.tajo.datum.Datum; |
| import org.apache.tajo.exception.TajoException; |
| import org.apache.tajo.exception.TajoInternalError; |
| import org.apache.tajo.plan.LogicalPlan; |
| import org.apache.tajo.plan.LogicalPlan.QueryBlock; |
| import org.apache.tajo.plan.LogicalPlanner; |
| import org.apache.tajo.plan.Target; |
| 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.rewrite.rules.FilterPushDownRule.FilterPushDownContext; |
| import org.apache.tajo.plan.rewrite.rules.IndexScanInfo.SimplePredicate; |
| import org.apache.tajo.plan.util.PlannerUtil; |
| import org.apache.tajo.plan.visitor.BasicLogicalPlanVisitor; |
| import org.apache.tajo.util.TUtil; |
| |
| import java.util.*; |
| |
| /** |
| * This rule tries to push down all filter conditions into logical nodes as lower as possible. |
| * It is likely to significantly reduces the intermediate data. |
| */ |
| public class FilterPushDownRule extends BasicLogicalPlanVisitor<FilterPushDownContext, LogicalNode> |
| implements LogicalPlanRewriteRule { |
| private final static Log LOG = LogFactory.getLog(FilterPushDownRule.class); |
| private static final String NAME = "FilterPushDown"; |
| |
| private CatalogService catalog; |
| |
| static class FilterPushDownContext { |
| Set<EvalNode> pushingDownFilters = new HashSet<>(); |
| LogicalPlanRewriteRuleContext rewriteRuleContext; |
| |
| public void clear() { |
| pushingDownFilters.clear(); |
| } |
| public void setFiltersTobePushed(Collection<EvalNode> workingEvals) { |
| this.pushingDownFilters.clear(); |
| this.pushingDownFilters.addAll(workingEvals); |
| } |
| public void addFiltersTobePushed(Collection<EvalNode> workingEvals) { |
| this.pushingDownFilters.addAll(workingEvals); |
| } |
| |
| public void setToOrigin(Map<EvalNode, EvalNode> evalMap) { |
| //evalMap: copy -> origin |
| List<EvalNode> origins = TUtil.newList(); |
| for (EvalNode eval : pushingDownFilters) { |
| EvalNode origin = evalMap.get(eval); |
| if (origin != null) { |
| origins.add(origin); |
| } |
| } |
| setFiltersTobePushed(origins); |
| } |
| } |
| |
| @Override |
| public String getName() { |
| return NAME; |
| } |
| |
| @Override |
| public boolean isEligible(LogicalPlanRewriteRuleContext context) { |
| for (LogicalPlan.QueryBlock block : context.getPlan().getQueryBlocks()) { |
| if (block.hasNode(NodeType.SELECTION) || block.hasNode(NodeType.JOIN)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| @Override |
| public LogicalPlan rewrite(LogicalPlanRewriteRuleContext rewriteRuleContext) throws TajoException { |
| /* |
| FilterPushDown rule: processing when visits each node |
| - If a target which is corresponding on a filter EvalNode's column is not FieldEval, do not PushDown. |
| - Replace filter EvalNode's column with child node's output column. |
| If there is no child node's output column, do not PushDown. |
| - When visit ScanNode, add filter eval to ScanNode's qual |
| - When visit GroupByNode, Find aggregation column in a filter EvalNode and |
| . If a parent is HavingNode, add filter eval to parent HavingNode. |
| . It not, create new HavingNode and set parent's child. |
| */ |
| FilterPushDownContext context = new FilterPushDownContext(); |
| context.rewriteRuleContext = rewriteRuleContext; |
| LogicalPlan plan = rewriteRuleContext.getPlan(); |
| catalog = rewriteRuleContext.getCatalog(); |
| for (LogicalPlan.QueryBlock block : plan.getQueryBlocks()) { |
| context.clear(); |
| this.visit(context, plan, block, block.getRoot(), new Stack<LogicalNode>()); |
| } |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("============================================="); |
| LOG.debug("FilterPushDown Optimized Query: \n" + plan.toString()); |
| LOG.debug("============================================="); |
| } |
| return plan; |
| } |
| |
| @Override |
| public LogicalNode visitFilter(FilterPushDownContext context, LogicalPlan plan, LogicalPlan.QueryBlock block, |
| SelectionNode selNode, Stack<LogicalNode> stack) throws TajoException { |
| context.pushingDownFilters.addAll(new HashSet<>(Arrays.asList(AlgebraicUtil.toConjunctiveNormalFormArray(selNode.getQual())))); |
| |
| stack.push(selNode); |
| visit(context, plan, block, selNode.getChild(), stack); |
| stack.pop(); |
| |
| if(context.pushingDownFilters.size() == 0) { |
| // remove the selection operator if there is no search condition after selection push. |
| LogicalNode node = stack.peek(); |
| if (node instanceof UnaryNode) { |
| UnaryNode unary = (UnaryNode) node; |
| unary.setChild(selNode.getChild()); |
| } else { |
| throw new TajoInternalError("The node must be an unary node"); |
| } |
| } else { // if there remain search conditions |
| |
| // check if it can be evaluated here |
| Set<EvalNode> matched = new HashSet<>(); |
| for (EvalNode eachEval : context.pushingDownFilters) { |
| if (LogicalPlanner.checkIfBeEvaluatedAtThis(eachEval, selNode)) { |
| matched.add(eachEval); |
| } |
| } |
| |
| // if there are search conditions which can be evaluated here, |
| // push down them and remove them from context.pushingDownFilters. |
| if (matched.size() > 0) { |
| selNode.setQual(AlgebraicUtil.createSingletonExprFromCNF(matched.toArray(new EvalNode[matched.size()]))); |
| context.pushingDownFilters.removeAll(matched); |
| } |
| } |
| |
| return selNode; |
| } |
| |
| @Override |
| public LogicalNode visitJoin(FilterPushDownContext context, LogicalPlan plan, LogicalPlan.QueryBlock block, |
| JoinNode joinNode, Stack<LogicalNode> stack) throws TajoException { |
| Set<EvalNode> onPredicates = new HashSet<>(); |
| if (joinNode.hasJoinQual()) { |
| onPredicates.addAll(new HashSet<>(Arrays.asList(AlgebraicUtil.toConjunctiveNormalFormArray(joinNode.getJoinQual())))); |
| } |
| // clear join qual |
| joinNode.clearJoinQual(); |
| |
| // we assume all the quals in pushingDownFilters as where predicates |
| Set<EvalNode> nonPushableQuals = extractNonPushableJoinQuals(plan, block, joinNode, onPredicates, |
| context.pushingDownFilters); |
| // add every predicate and remove non-pushable ones |
| context.pushingDownFilters.addAll(onPredicates); |
| context.pushingDownFilters.removeAll(nonPushableQuals); |
| |
| LogicalNode left = joinNode.getLeftChild(); |
| LogicalNode right = joinNode.getRightChild(); |
| |
| List<EvalNode> notMatched = TUtil.newList(); |
| // Join's input schema = right child output columns + left child output columns |
| Map<EvalNode, EvalNode> transformedMap = findCanPushdownAndTransform(context, block, joinNode, left, notMatched, |
| null, 0); |
| context.setFiltersTobePushed(transformedMap.keySet()); |
| visit(context, plan, block, left, stack); |
| |
| context.setToOrigin(transformedMap); |
| context.addFiltersTobePushed(notMatched); |
| |
| notMatched.clear(); |
| transformedMap = findCanPushdownAndTransform(context, block, joinNode, right, notMatched, null, |
| left.getOutSchema().size()); |
| context.setFiltersTobePushed(transformedMap.keySet()); |
| |
| visit(context, plan, block, right, stack); |
| |
| context.setToOrigin(transformedMap); |
| context.addFiltersTobePushed(notMatched); |
| |
| notMatched.clear(); |
| context.pushingDownFilters.addAll(nonPushableQuals); |
| List<EvalNode> matched = TUtil.newList(); |
| |
| // If the query involves a subquery, the stack can be empty. |
| // In this case, this join is the top most one within a query block. |
| boolean isTopMostJoin = stack.isEmpty() ? true : stack.peek().getType() != NodeType.JOIN; |
| |
| for (EvalNode evalNode : context.pushingDownFilters) { |
| // TODO: currently, non-equi theta join is not supported yet. |
| if (LogicalPlanner.isEvaluatableJoinQual(block, evalNode, joinNode, onPredicates.contains(evalNode), |
| isTopMostJoin)) { |
| matched.add(evalNode); |
| } |
| } |
| |
| EvalNode qual = null; |
| if (matched.size() > 1) { |
| // merged into one eval tree |
| qual = AlgebraicUtil.createSingletonExprFromCNF( |
| matched.toArray(new EvalNode[matched.size()])); |
| } else if (matched.size() == 1) { |
| // if the number of matched expr is one |
| qual = matched.get(0); |
| } |
| |
| if (qual != null) { |
| joinNode.setJoinQual(qual); |
| |
| if (joinNode.getJoinType() == JoinType.CROSS) { |
| joinNode.setJoinType(JoinType.INNER); |
| } |
| } |
| |
| context.pushingDownFilters.removeAll(matched); |
| |
| // The selection node for non-equi theta join conditions should be created here |
| // to process those conditions as early as possible. |
| // This should be removed after TAJO-742. |
| if (context.pushingDownFilters.size() > 0) { |
| List<EvalNode> nonEquiThetaJoinQuals = extractNonEquiThetaJoinQuals(context.pushingDownFilters, block, joinNode); |
| if (nonEquiThetaJoinQuals.size() > 0) { |
| SelectionNode selectionNode = createSelectionParentForNonEquiThetaJoinQuals(plan, block, stack, joinNode, |
| nonEquiThetaJoinQuals); |
| |
| context.pushingDownFilters.removeAll(nonEquiThetaJoinQuals); |
| return selectionNode; |
| } |
| } |
| |
| return joinNode; |
| } |
| |
| private SelectionNode createSelectionParentForNonEquiThetaJoinQuals(LogicalPlan plan, |
| QueryBlock block, |
| Stack<LogicalNode> stack, |
| JoinNode joinNode, |
| List<EvalNode> nonEquiThetaJoinQuals) { |
| SelectionNode selectionNode = plan.createNode(SelectionNode.class); |
| selectionNode.setInSchema(joinNode.getOutSchema()); |
| selectionNode.setOutSchema(joinNode.getOutSchema()); |
| selectionNode.setQual(AlgebraicUtil.createSingletonExprFromCNF(nonEquiThetaJoinQuals)); |
| block.registerNode(selectionNode); |
| |
| LogicalNode parent = stack.peek(); |
| if (parent instanceof UnaryNode) { |
| ((UnaryNode) parent).setChild(selectionNode); |
| } else if (parent instanceof BinaryNode) { |
| BinaryNode binaryParent = (BinaryNode) parent; |
| if (binaryParent.getLeftChild().getPID() == joinNode.getPID()) { |
| binaryParent.setLeftChild(selectionNode); |
| } else if (binaryParent.getRightChild().getPID() == joinNode.getPID()) { |
| binaryParent.setRightChild(selectionNode); |
| } |
| } else if (parent instanceof TableSubQueryNode) { |
| ((TableSubQueryNode) parent).setSubQuery(selectionNode); |
| } |
| |
| selectionNode.setChild(joinNode); |
| return selectionNode; |
| } |
| |
| private static Set<EvalNode> extractNonPushableJoinQuals(final LogicalPlan plan, |
| final LogicalPlan.QueryBlock block, |
| final JoinNode joinNode, |
| final Set<EvalNode> onPredicates, |
| final Set<EvalNode> wherePredicates) |
| throws TajoException { |
| Set<EvalNode> nonPushableQuals = new HashSet<>(); |
| // TODO: non-equi theta join quals must not be pushed until TAJO-742 is resolved. |
| nonPushableQuals.addAll(extractNonEquiThetaJoinQuals(wherePredicates, block, joinNode)); |
| nonPushableQuals.addAll(extractNonEquiThetaJoinQuals(onPredicates, block, joinNode)); |
| |
| // for outer joins |
| if (PlannerUtil.isOuterJoinType(joinNode.getJoinType())) { |
| nonPushableQuals.addAll(extractNonPushableOuterJoinQuals(plan, onPredicates, wherePredicates, joinNode)); |
| } |
| return nonPushableQuals; |
| } |
| |
| /** |
| * For outer joins, pushable predicates can be decided based on their locations in the SQL and types of referencing |
| * relations. |
| * |
| * <h3>Table types</h3> |
| * <ul> |
| * <li>Preserved Row table : The preserved row table refers to the table that preserves rows when there is no match |
| * in the join operation. Therefore, all rows from the preserved row table that qualify against the WHERE clause |
| * will be returned, regardless of whether there is a matched row in the join. For a left/right table, the preserved |
| * row table is the left/right table. For a full outer join, both tables are preserved row tables.</li> |
| * <li>Null Supplying table : The NULL-supplying table supplies NULLs when there is an unmatched row. Any column |
| * from the NULL-supplying table referred to in the SELECT list or subsequent WHERE or ON clause will contain NULL |
| * if there was no match in the join operation. For a left/right outer join, the NULL-supplying |
| * table is the right/left table. For a full outer join, both tables are NULL-supplying |
| * table. In a full outer join, both tables can preserve rows, and also can supply NULLs. This is significant, |
| * because there are rules that apply to purely preserved row tables that do not apply if the table can also supply |
| * NULLs.</li> |
| * </ul> |
| * |
| * <h3>Predicate types</h3> |
| * <ul> |
| * <li>During Join predicate : A predicate that is in the JOIN ON clause.</li> |
| * <li>After Join predicate : A predicate that is in the WHERE clause.</li> |
| * </ul> |
| * |
| * <h3>Predicate Pushdown Rules</h3> |
| * <ol> |
| * <li>During Join predicates cannot be pushed past Preserved Row tables.</li> |
| * <li>After Join predicates cannot be pushed past Null Supplying tables.</li> |
| * </ol> |
| */ |
| private static Set<EvalNode> extractNonPushableOuterJoinQuals(final LogicalPlan plan, |
| final Set<EvalNode> onPredicates, |
| final Set<EvalNode> wherePredicates, |
| final JoinNode joinNode) throws TajoException { |
| Set<String> nullSupplyingTableNameSet = new HashSet<>(); |
| Set<String> preservedTableNameSet = new HashSet<>(); |
| String leftRelation = PlannerUtil.getTopRelationInLineage(plan, joinNode.getLeftChild()); |
| String rightRelation = PlannerUtil.getTopRelationInLineage(plan, joinNode.getRightChild()); |
| |
| if (joinNode.getJoinType() == JoinType.LEFT_OUTER) { |
| nullSupplyingTableNameSet.add(rightRelation); |
| preservedTableNameSet.add(leftRelation); |
| } else if (joinNode.getJoinType() == JoinType.RIGHT_OUTER) { |
| nullSupplyingTableNameSet.add(leftRelation); |
| preservedTableNameSet.add(rightRelation); |
| } else { |
| // full outer join |
| preservedTableNameSet.add(leftRelation); |
| preservedTableNameSet.add(rightRelation); |
| nullSupplyingTableNameSet.add(leftRelation); |
| nullSupplyingTableNameSet.add(rightRelation); |
| } |
| |
| Set<EvalNode> nonPushableQuals = new HashSet<>(); |
| for (EvalNode eachQual : onPredicates) { |
| for (String relName : preservedTableNameSet) { |
| if (isEvalNeedRelation(eachQual, relName)) { |
| nonPushableQuals.add(eachQual); |
| } |
| } |
| } |
| |
| for (EvalNode eachQual : wherePredicates) { |
| for (String relName : nullSupplyingTableNameSet) { |
| if (isEvalNeedRelation(eachQual, relName)) { |
| nonPushableQuals.add(eachQual); |
| } |
| } |
| } |
| |
| return nonPushableQuals; |
| } |
| |
| private static boolean isEvalNeedRelation(final EvalNode evalNode, final String relationName) { |
| Set<Column> columns = EvalTreeUtil.findUniqueColumns(evalNode); |
| for (Column column : columns) { |
| if (isColumnFromRelation(column, relationName)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| private static boolean isColumnFromRelation(final Column column, final String relationName) { |
| if (relationName.equals(column.getQualifier())) { |
| return true; |
| } |
| return false; |
| } |
| |
| private static boolean isNonEquiThetaJoinQual(final LogicalPlan.QueryBlock block, |
| final JoinNode joinNode, |
| final EvalNode evalNode) { |
| if (EvalTreeUtil.isJoinQual(block, joinNode.getLeftChild().getOutSchema(), |
| joinNode.getRightChild().getOutSchema(), evalNode, true) && |
| evalNode.getType() != EvalType.EQUAL) { |
| return true; |
| } else { |
| return false; |
| } |
| } |
| |
| private static List<EvalNode> extractNonEquiThetaJoinQuals(final Set<EvalNode> predicates, |
| final LogicalPlan.QueryBlock block, |
| final JoinNode joinNode) { |
| List<EvalNode> nonEquiThetaJoinQuals = TUtil.newList(); |
| for (EvalNode eachEval: predicates) { |
| if (isNonEquiThetaJoinQual(block, joinNode, eachEval)) { |
| nonEquiThetaJoinQuals.add(eachEval); |
| } |
| } |
| return nonEquiThetaJoinQuals; |
| } |
| |
| private Map<EvalNode, EvalNode> transformEvalsWidthByPassNode( |
| Collection<EvalNode> originEvals, LogicalPlan plan, |
| LogicalPlan.QueryBlock block, |
| LogicalNode node, LogicalNode childNode) throws TajoException { |
| // transformed -> pushingDownFilters |
| Map<EvalNode, EvalNode> transformedMap = new HashMap<>(); |
| |
| if (originEvals.isEmpty()) { |
| return transformedMap; |
| } |
| |
| if (node.getType() == NodeType.UNION) { |
| // If node is union, All eval's columns are simple name and matched with child's output schema. |
| Schema childOutSchema = childNode.getOutSchema(); |
| for (EvalNode eval : originEvals) { |
| EvalNode copy; |
| try { |
| copy = (EvalNode) eval.clone(); |
| } catch (CloneNotSupportedException e) { |
| throw new TajoInternalError(e); |
| } |
| |
| Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy); |
| for (Column c : columns) { |
| Column column = childOutSchema.getColumn(c.getSimpleName()); |
| if (column == null) { |
| throw new TajoInternalError( |
| "Invalid Filter PushDown on SubQuery: No such a corresponding column '" |
| + c.getQualifiedName() + " for FilterPushDown(" + eval + "), " + |
| "(PID=" + node.getPID() + ", Child=" + childNode.getPID() + ")"); |
| } |
| EvalTreeUtil.changeColumnRef(copy, c.getSimpleName(), column.getQualifiedName()); |
| } |
| |
| transformedMap.put(copy, eval); |
| } |
| return transformedMap; |
| } |
| |
| if (childNode.getType() == NodeType.UNION) { |
| // If child is union, remove qualifier from eval's column |
| for (EvalNode eval : originEvals) { |
| EvalNode copy; |
| try { |
| copy = (EvalNode) eval.clone(); |
| } catch (CloneNotSupportedException e) { |
| throw new TajoInternalError(e); |
| } |
| |
| Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy); |
| for (Column c : columns) { |
| if (c.hasQualifier()) { |
| EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), c.getSimpleName()); |
| } |
| } |
| |
| transformedMap.put(copy, eval); |
| } |
| |
| return transformedMap; |
| } |
| |
| // node in column -> child out column |
| Map<String, String> columnMap = new HashMap<>(); |
| |
| for (int i = 0; i < node.getInSchema().size(); i++) { |
| String inColumnName = node.getInSchema().getColumn(i).getQualifiedName(); |
| Column childOutColumn = childNode.getOutSchema().getColumn(i); |
| columnMap.put(inColumnName, childOutColumn.getQualifiedName()); |
| } |
| |
| // Rename from upper block's one to lower block's one |
| for (EvalNode matchedEval : originEvals) { |
| EvalNode copy; |
| try { |
| copy = (EvalNode) matchedEval.clone(); |
| } catch (CloneNotSupportedException e) { |
| throw new TajoInternalError(e); |
| } |
| |
| Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy); |
| boolean allMatched = true; |
| for (Column c : columns) { |
| if (columnMap.containsKey(c.getQualifiedName())) { |
| EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), columnMap.get(c.getQualifiedName())); |
| } else { |
| if (childNode.getType() == NodeType.GROUP_BY) { |
| if (((GroupbyNode) childNode).isAggregationColumn(c.getSimpleName())) { |
| allMatched = false; |
| break; |
| } |
| } else { |
| throw new TajoInternalError( |
| "Invalid Filter PushDown on SubQuery: No such a corresponding column '" |
| + c.getQualifiedName() + " for FilterPushDown(" + matchedEval + "), " + |
| "(PID=" + node.getPID() + ", Child=" + childNode.getPID() + ")" |
| ); |
| } |
| } |
| } |
| if (allMatched) { |
| transformedMap.put(copy, matchedEval); |
| } |
| } |
| |
| return transformedMap; |
| } |
| |
| @Override |
| public LogicalNode visitTableSubQuery(FilterPushDownContext context, LogicalPlan plan, LogicalPlan.QueryBlock block, |
| TableSubQueryNode node, Stack<LogicalNode> stack) throws TajoException { |
| List<EvalNode> matched = TUtil.newList(); |
| List<EvalNode> unmatched = TUtil.newList(); |
| for (EvalNode eval : context.pushingDownFilters) { |
| if (LogicalPlanner.checkIfBeEvaluatedAtRelation(block, eval, node)) { |
| matched.add(eval); |
| } else { |
| unmatched.add(eval); |
| } |
| } |
| |
| // transformed -> pushingDownFilters |
| Map<EvalNode, EvalNode> transformedMap = |
| transformEvalsWidthByPassNode(matched, plan, block, node, node.getSubQuery()); |
| context.setFiltersTobePushed(transformedMap.keySet()); |
| visit(context, plan, plan.getBlock(node.getSubQuery())); |
| context.setToOrigin(transformedMap); |
| context.addFiltersTobePushed(unmatched); |
| |
| return node; |
| } |
| |
| @Override |
| public LogicalNode visitUnion(FilterPushDownContext context, LogicalPlan plan, |
| LogicalPlan.QueryBlock block, UnionNode unionNode, |
| Stack<LogicalNode> stack) throws TajoException { |
| LogicalNode leftNode = unionNode.getLeftChild(); |
| |
| List<EvalNode> origins = TUtil.newList(context.pushingDownFilters); |
| |
| // transformed -> pushingDownFilters |
| Map<EvalNode, EvalNode> transformedMap = transformEvalsWidthByPassNode(origins, plan, block, unionNode, leftNode); |
| context.setFiltersTobePushed(transformedMap.keySet()); |
| visit(context, plan, plan.getBlock(leftNode)); |
| |
| if (!context.pushingDownFilters.isEmpty()) { |
| errorFilterPushDown(plan, leftNode, context); |
| } |
| |
| LogicalNode rightNode = unionNode.getRightChild(); |
| transformedMap = transformEvalsWidthByPassNode(origins, plan, block, unionNode, rightNode); |
| context.setFiltersTobePushed(transformedMap.keySet()); |
| visit(context, plan, plan.getBlock(rightNode), rightNode, stack); |
| |
| if (!context.pushingDownFilters.isEmpty()) { |
| errorFilterPushDown(plan, rightNode, context); |
| } |
| |
| // notify all filter matched to upper |
| context.pushingDownFilters.clear(); |
| return unionNode; |
| } |
| |
| @Override |
| public LogicalNode visitProjection(FilterPushDownContext context, |
| LogicalPlan plan, |
| LogicalPlan.QueryBlock block, |
| ProjectionNode projectionNode, |
| Stack<LogicalNode> stack) throws TajoException { |
| LogicalNode childNode = projectionNode.getChild(); |
| |
| List<EvalNode> notMatched = TUtil.newList(); |
| |
| //copy -> origin |
| BiMap<EvalNode, EvalNode> transformedMap = findCanPushdownAndTransform( |
| context, block,projectionNode, childNode, notMatched, null, 0); |
| |
| context.setFiltersTobePushed(transformedMap.keySet()); |
| |
| stack.push(projectionNode); |
| visit(context, plan, plan.getBlock(childNode), childNode, stack); |
| stack.pop(); |
| |
| // find not matched after visiting child |
| for (EvalNode eval: context.pushingDownFilters) { |
| notMatched.add(transformedMap.get(eval)); |
| } |
| |
| EvalNode qual = null; |
| if (notMatched.size() > 1) { |
| // merged into one eval tree |
| qual = AlgebraicUtil.createSingletonExprFromCNF(notMatched.toArray(new EvalNode[notMatched.size()])); |
| } else if (notMatched.size() == 1) { |
| // if the number of matched expr is one |
| qual = notMatched.get(0); |
| } |
| |
| // If there is not matched node add SelectionNode and clear context.pushingDownFilters |
| if (qual != null && LogicalPlanner.checkIfBeEvaluatedAtThis(qual, projectionNode)) { |
| SelectionNode selectionNode = plan.createNode(SelectionNode.class); |
| selectionNode.setInSchema(childNode.getOutSchema()); |
| selectionNode.setOutSchema(childNode.getOutSchema()); |
| selectionNode.setQual(qual); |
| block.registerNode(selectionNode); |
| |
| projectionNode.setChild(selectionNode); |
| selectionNode.setChild(childNode); |
| |
| // clean all remain filters because all conditions are merged into a qual |
| context.pushingDownFilters.clear(); |
| } |
| |
| // if there are remain filters, recover the original names and give back to the upper query block. |
| if (context.pushingDownFilters.size() > 0) { |
| ImmutableSet<EvalNode> copy = ImmutableSet.copyOf(context.pushingDownFilters); |
| context.pushingDownFilters.clear(); |
| context.pushingDownFilters.addAll(reverseTransform(transformedMap, copy)); |
| } |
| |
| return projectionNode; |
| } |
| |
| private Collection<EvalNode> reverseTransform(BiMap<EvalNode, EvalNode> map, Set<EvalNode> remainFilters) { |
| Set<EvalNode> reversed = new HashSet<>(); |
| for (EvalNode evalNode : remainFilters) { |
| reversed.add(map.get(evalNode)); |
| } |
| return reversed; |
| } |
| |
| private BiMap<EvalNode, EvalNode> findCanPushdownAndTransform( |
| FilterPushDownContext context, LogicalPlan.QueryBlock block, Projectable node, |
| LogicalNode childNode, List<EvalNode> notMatched, |
| Set<String> partitionColumns, int columnOffset) throws TajoException { |
| // canonical name -> target |
| Map<String, Target> nodeTargetMap = new HashMap<>(); |
| for (Target target : node.getTargets()) { |
| nodeTargetMap.put(target.getCanonicalName(), target); |
| } |
| |
| // copy -> origin |
| BiMap<EvalNode, EvalNode> matched = HashBiMap.create(); |
| |
| for (EvalNode eval : context.pushingDownFilters) { |
| // If all column is field eval, can push down. |
| Set<Column> evalColumns = EvalTreeUtil.findUniqueColumns(eval); |
| boolean columnMatched = true; |
| for (Column c : evalColumns) { |
| Target target = nodeTargetMap.get(c.getQualifiedName()); |
| if (target == null) { |
| columnMatched = false; |
| break; |
| } |
| if (target.getEvalTree().getType() != EvalType.FIELD) { |
| columnMatched = false; |
| break; |
| } |
| } |
| |
| if (columnMatched) { |
| // transform eval column to child's output column |
| EvalNode copyEvalNode = transformEval(node, childNode, eval, nodeTargetMap, partitionColumns, columnOffset); |
| if (copyEvalNode != null) { |
| matched.put(copyEvalNode, eval); |
| } else { |
| notMatched.add(eval); |
| } |
| } else { |
| notMatched.add(eval); |
| } |
| } |
| |
| return matched; |
| } |
| |
| private EvalNode transformEval(Projectable node, LogicalNode childNode, EvalNode origin, |
| Map<String, Target> targetMap, Set<String> partitionColumns, |
| int columnOffset) throws TajoException { |
| Schema outputSchema = childNode != null ? childNode.getOutSchema() : node.getInSchema(); |
| EvalNode copy; |
| try { |
| copy = (EvalNode) origin.clone(); |
| } catch (CloneNotSupportedException e) { |
| throw new TajoInternalError(e); |
| } |
| Set<Column> columns = EvalTreeUtil.findUniqueColumns(copy); |
| for (Column c: columns) { |
| Target target = targetMap.get(c.getQualifiedName()); |
| if (target == null) { |
| throw new TajoInternalError( |
| "Invalid Filter PushDown: No such a corresponding target '" |
| + c.getQualifiedName() + " for FilterPushDown(" + origin + "), " + |
| "(PID=" + node.getPID() + ")" |
| ); |
| } |
| EvalNode targetEvalNode = target.getEvalTree(); |
| if (targetEvalNode.getType() != EvalType.FIELD) { |
| throw new TajoInternalError( |
| "Invalid Filter PushDown: '" + c.getQualifiedName() + "' target is not FieldEval " + |
| "(PID=" + node.getPID() + ")" |
| ); |
| } |
| |
| FieldEval fieldEval = (FieldEval)targetEvalNode; |
| Column targetInputColumn = fieldEval.getColumnRef(); |
| |
| int index; |
| if (targetInputColumn.hasQualifier()) { |
| index = node.getInSchema().getColumnId(targetInputColumn.getQualifiedName()); |
| } else { |
| index = node.getInSchema().getColumnIdByName(targetInputColumn.getQualifiedName()); |
| } |
| if (columnOffset > 0) { |
| index = index - columnOffset; |
| } |
| if (index < 0 || index >= outputSchema.size()) { |
| if (partitionColumns != null && !partitionColumns.isEmpty() && node instanceof ScanNode) { |
| ScanNode scanNode = (ScanNode)node; |
| boolean isPartitionColumn = false; |
| if (CatalogUtil.isFQColumnName(partitionColumns.iterator().next())) { |
| isPartitionColumn = partitionColumns.contains( |
| CatalogUtil.buildFQName(scanNode.getTableName(), c.getSimpleName())); |
| } else { |
| isPartitionColumn = partitionColumns.contains(c.getSimpleName()); |
| } |
| if (isPartitionColumn) { |
| EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), |
| scanNode.getCanonicalName() + "." + c.getSimpleName()); |
| } else { |
| return null; |
| } |
| } else { |
| return null; |
| } |
| } else { |
| Column outputColumn = outputSchema.getColumn(index); |
| EvalTreeUtil.changeColumnRef(copy, c.getQualifiedName(), outputColumn.getQualifiedName()); |
| } |
| } |
| |
| return copy; |
| } |
| |
| /** |
| * Find aggregation columns in filter eval and add having clause or add HavingNode. |
| * @param context |
| * @param plan |
| * @param block |
| * @param parentNode If null, having is parent |
| * @param havingNode If null, projection is parent |
| * @param groupByNode |
| * @return matched origin eval |
| * @throws TajoException |
| */ |
| private List<EvalNode> addHavingNode(FilterPushDownContext context, LogicalPlan plan, |
| LogicalPlan.QueryBlock block, |
| UnaryNode parentNode, |
| HavingNode havingNode, |
| GroupbyNode groupByNode) throws TajoException { |
| // find aggregation column |
| Set<Column> groupingColumns = new HashSet<>(Arrays.asList(groupByNode.getGroupingColumns())); |
| Set<String> aggrFunctionOutColumns = new HashSet<>(); |
| for (Column column : groupByNode.getOutSchema().getRootColumns()) { |
| if (!groupingColumns.contains(column)) { |
| aggrFunctionOutColumns.add(column.getQualifiedName()); |
| } |
| } |
| |
| List<EvalNode> aggrEvalOrigins = TUtil.newList(); |
| List<EvalNode> aggrEvals = TUtil.newList(); |
| |
| for (EvalNode eval : context.pushingDownFilters) { |
| EvalNode copy = null; |
| try { |
| copy = (EvalNode)eval.clone(); |
| } catch (CloneNotSupportedException e) { |
| } |
| boolean isEvalAggrFunction = false; |
| for (Column evalColumn : EvalTreeUtil.findUniqueColumns(copy)) { |
| if (aggrFunctionOutColumns.contains(evalColumn.getSimpleName())) { |
| EvalTreeUtil.changeColumnRef(copy, evalColumn.getQualifiedName(), evalColumn.getSimpleName()); |
| isEvalAggrFunction = true; |
| break; |
| } |
| } |
| if (groupByNode.getInSchema().containsAll(EvalTreeUtil.findUniqueColumns(copy))) { |
| isEvalAggrFunction = true; |
| } |
| if (isEvalAggrFunction) { |
| aggrEvals.add(copy); |
| aggrEvalOrigins.add(eval); |
| } |
| } |
| |
| if (aggrEvals.isEmpty()) { |
| return aggrEvalOrigins; |
| } |
| |
| // transform |
| HavingNode workingHavingNode; |
| if (havingNode != null) { |
| workingHavingNode = havingNode; |
| aggrEvals.add(havingNode.getQual()); |
| } else { |
| workingHavingNode = plan.createNode(HavingNode.class); |
| block.registerNode(workingHavingNode); |
| parentNode.setChild(workingHavingNode); |
| workingHavingNode.setChild(groupByNode); |
| } |
| |
| EvalNode qual = null; |
| if (aggrEvals.size() > 1) { |
| // merged into one eval tree |
| qual = AlgebraicUtil.createSingletonExprFromCNF(aggrEvals.toArray(new EvalNode[aggrEvals.size()])); |
| } else if (aggrEvals.size() == 1) { |
| // if the number of matched expr is one |
| qual = aggrEvals.get(0); |
| } |
| |
| // If there is not matched node add SelectionNode and clear context.pushingDownFilters |
| if (qual != null) { |
| workingHavingNode.setQual(qual); |
| } |
| |
| return aggrEvalOrigins; |
| } |
| |
| @Override |
| public LogicalNode visitWindowAgg(FilterPushDownContext context, LogicalPlan plan, |
| LogicalPlan.QueryBlock block, WindowAggNode winAggNode, |
| Stack<LogicalNode> stack) throws TajoException { |
| stack.push(winAggNode); |
| super.visitWindowAgg(context, plan, block, winAggNode, stack); |
| stack.pop(); |
| return winAggNode; |
| } |
| |
| |
| @Override |
| public LogicalNode visitGroupBy(FilterPushDownContext context, LogicalPlan plan, |
| LogicalPlan.QueryBlock block, GroupbyNode groupbyNode, |
| Stack<LogicalNode> stack) throws TajoException { |
| LogicalNode parentNode = stack.peek(); |
| List<EvalNode> aggrEvals; |
| if (parentNode.getType() == NodeType.HAVING) { |
| aggrEvals = addHavingNode(context, plan, block, null, (HavingNode)parentNode, groupbyNode); |
| } else { |
| aggrEvals = addHavingNode(context, plan, block, (UnaryNode)parentNode, null, groupbyNode); |
| } |
| |
| if (aggrEvals != null) { |
| // remove aggregation eval from conext |
| context.pushingDownFilters.removeAll(aggrEvals); |
| } |
| |
| List<EvalNode> notMatched = TUtil.newList(); |
| // transform |
| Map<EvalNode, EvalNode> transformed = |
| findCanPushdownAndTransform(context, block, groupbyNode,groupbyNode.getChild(), notMatched, null, 0); |
| |
| context.setFiltersTobePushed(transformed.keySet()); |
| LogicalNode current = super.visitGroupBy(context, plan, block, groupbyNode, stack); |
| |
| context.setToOrigin(transformed); |
| context.addFiltersTobePushed(notMatched); |
| |
| return current; |
| } |
| |
| @Override |
| public LogicalNode visitScan(FilterPushDownContext context, LogicalPlan plan, |
| LogicalPlan.QueryBlock block, final ScanNode scanNode, |
| Stack<LogicalNode> stack) throws TajoException { |
| List<EvalNode> matched = TUtil.newList(); |
| |
| // find partition column and check matching |
| Set<String> partitionColumns = new HashSet<>(); |
| TableDesc table = scanNode.getTableDesc(); |
| boolean hasQualifiedName = false; |
| if (table.hasPartition()) { |
| for (Column c: table.getPartitionMethod().getExpressionSchema().getRootColumns()) { |
| partitionColumns.add(c.getQualifiedName()); |
| hasQualifiedName = c.hasQualifier(); |
| } |
| } |
| Set<EvalNode> partitionEvals = new HashSet<>(); |
| for (EvalNode eval : context.pushingDownFilters) { |
| if (table.hasPartition()) { |
| Set<Column> columns = EvalTreeUtil.findUniqueColumns(eval); |
| if (columns.size() != 1) { |
| continue; |
| } |
| Column column = columns.iterator().next(); |
| |
| // If catalog runs with HiveCatalogStore, partition column is a qualified name |
| // Else partition column is a simple name |
| boolean isPartitionColumn = false; |
| if (hasQualifiedName) { |
| isPartitionColumn = partitionColumns.contains(CatalogUtil.buildFQName(table.getName(), column.getSimpleName())); |
| } else { |
| isPartitionColumn = partitionColumns.contains(column.getSimpleName()); |
| } |
| if (isPartitionColumn) { |
| EvalNode copy; |
| try { |
| copy = (EvalNode) eval.clone(); |
| } catch (CloneNotSupportedException e) { |
| throw new TajoInternalError(e); |
| } |
| EvalTreeUtil.changeColumnRef(copy, column.getQualifiedName(), |
| scanNode.getCanonicalName() + "." + column.getSimpleName()); |
| matched.add(copy); |
| partitionEvals.add(eval); |
| } |
| } |
| } |
| |
| context.pushingDownFilters.removeAll(partitionEvals); |
| |
| List<EvalNode> notMatched = TUtil.newList(); |
| |
| // transform |
| Map<EvalNode, EvalNode> transformed = |
| findCanPushdownAndTransform(context, block, scanNode, null, notMatched, partitionColumns, 0); |
| |
| for (EvalNode eval : transformed.keySet()) { |
| if (LogicalPlanner.checkIfBeEvaluatedAtRelation(block, eval, scanNode)) { |
| matched.add(eval); |
| } |
| } |
| |
| EvalNode qual = null; |
| if (matched.size() > 1) { |
| // merged into one eval tree |
| qual = AlgebraicUtil.createSingletonExprFromCNF( |
| matched.toArray(new EvalNode[matched.size()])); |
| } else if (matched.size() == 1) { |
| // if the number of matched expr is one |
| qual = matched.iterator().next(); |
| } |
| |
| block.addAccessPath(scanNode, new SeqScanInfo(table)); |
| if (qual != null) { // if a matched qual exists |
| scanNode.setQual(qual); |
| |
| // Index path can be identified only after filters are pushed into each scan. |
| if(context.rewriteRuleContext.getQueryContext().getBool(SessionVars.INDEX_ENABLED)) { |
| String databaseName, tableName; |
| databaseName = CatalogUtil.extractQualifier(table.getName()); |
| tableName = CatalogUtil.extractSimpleName(table.getName()); |
| Set<Predicate> predicates = new HashSet<>(); |
| for (EvalNode eval : PlannerUtil.getAllEqualEvals(qual)) { |
| BinaryEval binaryEval = (BinaryEval) eval; |
| // TODO: consider more complex predicates |
| if (binaryEval.getLeftExpr().getType() == EvalType.FIELD && |
| binaryEval.getRightExpr().getType() == EvalType.CONST) { |
| predicates.add(new Predicate(binaryEval.getType(), |
| ((FieldEval) binaryEval.getLeftExpr()).getColumnRef(), |
| ((ConstEval) binaryEval.getRightExpr()).getValue())); |
| } else if (binaryEval.getLeftExpr().getType() == EvalType.CONST && |
| binaryEval.getRightExpr().getType() == EvalType.FIELD) { |
| predicates.add(new Predicate(binaryEval.getType(), |
| ((FieldEval) binaryEval.getRightExpr()).getColumnRef(), |
| ((ConstEval) binaryEval.getLeftExpr()).getValue())); |
| } |
| } |
| |
| // for every subset of the set of columns, find all matched index paths |
| for (Set<Predicate> subset : Sets.powerSet(predicates)) { |
| if (subset.size() == 0) |
| continue; |
| Column[] columns = extractColumns(subset); |
| if (catalog.existIndexByColumns(databaseName, tableName, columns)) { |
| IndexDesc indexDesc = catalog.getIndexByColumns(databaseName, tableName, columns); |
| block.addAccessPath(scanNode, new IndexScanInfo( |
| table.getStats(), indexDesc, getSimplePredicates(indexDesc, subset))); |
| } |
| } |
| } |
| } |
| |
| for (EvalNode matchedEval: matched) { |
| transformed.remove(matchedEval); |
| } |
| |
| context.setToOrigin(transformed); |
| context.addFiltersTobePushed(notMatched); |
| |
| return scanNode; |
| } |
| |
| private static class Predicate { |
| Column column; |
| Datum value; |
| EvalType evalType; |
| |
| public Predicate(EvalType evalType, Column column, Datum value) { |
| this.evalType = evalType; |
| this.column = column; |
| this.value = value; |
| } |
| } |
| |
| private static SimplePredicate[] getSimplePredicates(IndexDesc desc, Set<Predicate> predicates) { |
| SimplePredicate[] simplePredicates = new SimplePredicate[predicates.size()]; |
| Map<Column, Datum> colToValue = new HashMap<>(); |
| for (Predicate predicate : predicates) { |
| colToValue.put(predicate.column, predicate.value); |
| } |
| SortSpec [] keySortSpecs = desc.getKeySortSpecs(); |
| for (int i = 0; i < keySortSpecs.length; i++) { |
| simplePredicates[i] = new SimplePredicate(keySortSpecs[i], |
| colToValue.get(keySortSpecs[i].getSortKey())); |
| } |
| return simplePredicates; |
| } |
| |
| private static Column[] extractColumns(Set<Predicate> predicates) { |
| Column[] columns = new Column[predicates.size()]; |
| int i = 0; |
| for (Predicate p : predicates) { |
| columns[i++] = p.column; |
| } |
| return columns; |
| } |
| |
| private void errorFilterPushDown(LogicalPlan plan, LogicalNode node, |
| FilterPushDownContext context) { |
| String prefix = ""; |
| StringBuilder notMatchedNodeStrBuilder = new StringBuilder(); |
| for (EvalNode notMatchedNode: context.pushingDownFilters) { |
| notMatchedNodeStrBuilder.append(prefix).append(notMatchedNode.toString()); |
| if (prefix.isEmpty()) { |
| prefix = ", "; |
| } |
| } |
| throw new TajoInternalError("FilterPushDown failed cause some filters not matched: " |
| + notMatchedNodeStrBuilder.toString() + "\n" + |
| "Error node: " + node.getPlanString() + "\n" + plan.toString()); |
| } |
| } |
