You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2014/01/27 23:15:56 UTC
[40/51] [partial] Initial commit of master branch from github
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
new file mode 100644
index 0000000..4bbb357
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java
@@ -0,0 +1,1133 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+import static org.apache.phoenix.schema.SaltingUtil.SALTING_COLUMN;
+
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.expression.AndExpression;
+import org.apache.phoenix.expression.CoerceExpression;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.join.ScanProjector;
+import org.apache.phoenix.parse.AliasedNode;
+import org.apache.phoenix.parse.AndParseNode;
+import org.apache.phoenix.parse.BetweenParseNode;
+import org.apache.phoenix.parse.BindTableNode;
+import org.apache.phoenix.parse.CaseParseNode;
+import org.apache.phoenix.parse.CastParseNode;
+import org.apache.phoenix.parse.ColumnParseNode;
+import org.apache.phoenix.parse.ComparisonParseNode;
+import org.apache.phoenix.parse.ConcreteTableNode;
+import org.apache.phoenix.parse.DerivedTableNode;
+import org.apache.phoenix.parse.EqualParseNode;
+import org.apache.phoenix.parse.FunctionParseNode;
+import org.apache.phoenix.parse.InListParseNode;
+import org.apache.phoenix.parse.IsNullParseNode;
+import org.apache.phoenix.parse.JoinTableNode;
+import org.apache.phoenix.parse.JoinTableNode.JoinType;
+import org.apache.phoenix.parse.LikeParseNode;
+import org.apache.phoenix.parse.NamedTableNode;
+import org.apache.phoenix.parse.NotParseNode;
+import org.apache.phoenix.parse.OrParseNode;
+import org.apache.phoenix.parse.OrderByNode;
+import org.apache.phoenix.parse.ParseNode;
+import org.apache.phoenix.parse.ParseNodeFactory;
+import org.apache.phoenix.parse.SelectStatement;
+import org.apache.phoenix.parse.StatelessTraverseAllParseNodeVisitor;
+import org.apache.phoenix.parse.TableName;
+import org.apache.phoenix.parse.TableNode;
+import org.apache.phoenix.parse.TableNodeVisitor;
+import org.apache.phoenix.parse.TraverseNoParseNodeVisitor;
+import org.apache.phoenix.parse.WildcardParseNode;
+import org.apache.phoenix.schema.AmbiguousColumnException;
+import org.apache.phoenix.schema.ColumnNotFoundException;
+import org.apache.phoenix.schema.ColumnRef;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PColumnImpl;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.schema.PName;
+import org.apache.phoenix.schema.PNameFactory;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableImpl;
+import org.apache.phoenix.schema.PTableType;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.util.SchemaUtil;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ListMultimap;
+import com.google.common.collect.Lists;
+
+
+public class JoinCompiler {
+
+ public enum ColumnRefType {
+ PREFILTER,
+ JOINLOCAL,
+ GENERAL,
+ }
+
+ public static class JoinSpec {
+ private TableRef mainTable;
+ private List<AliasedNode> select; // all basic nodes related to mainTable, no aggregation.
+ private List<ParseNode> preFilters;
+ private List<ParseNode> postFilters;
+ private List<JoinTable> joinTables;
+ private Map<ColumnRef, ColumnRefType> columnRefs;
+
+ private JoinSpec(SelectStatement statement, ColumnResolver resolver) throws SQLException {
+ List<AliasedNode> selectList = statement.getSelect();
+ List<TableNode> tableNodes = statement.getFrom();
+ assert (tableNodes.size() > 1);
+ Iterator<TableNode> iter = tableNodes.iterator();
+ Iterator<TableRef> tableRefIter = resolver.getTables().iterator();
+ iter.next();
+ this.mainTable = tableRefIter.next();
+ this.select = extractFromSelect(selectList, mainTable, resolver);
+ this.joinTables = new ArrayList<JoinTable>(tableNodes.size() - 1);
+ this.preFilters = new ArrayList<ParseNode>();
+ this.postFilters = new ArrayList<ParseNode>();
+ ColumnParseNodeVisitor generalRefVisitor = new ColumnParseNodeVisitor(resolver);
+ ColumnParseNodeVisitor joinLocalRefVisitor = new ColumnParseNodeVisitor(resolver);
+ ColumnParseNodeVisitor prefilterRefVisitor = new ColumnParseNodeVisitor(resolver);
+ boolean hasRightJoin = false;
+ TableNode tableNode = null;
+ while (iter.hasNext()) {
+ tableNode = iter.next();
+ if (!(tableNode instanceof JoinTableNode))
+ throw new SQLFeatureNotSupportedException("Full joins not supported.");
+ JoinTableNode joinTableNode = (JoinTableNode) tableNode;
+ JoinTable joinTable = new JoinTable(joinTableNode, tableRefIter.next(), selectList, resolver);
+ joinTables.add(joinTable);
+ for (ParseNode prefilter : joinTable.preFilters) {
+ prefilter.accept(prefilterRefVisitor);
+ }
+ for (ParseNode condition : joinTable.conditions) {
+ ComparisonParseNode comparisonNode = (ComparisonParseNode) condition;
+ comparisonNode.getLHS().accept(generalRefVisitor);
+ comparisonNode.getRHS().accept(joinLocalRefVisitor);
+ }
+ if (joinTable.getType() == JoinType.Right) {
+ hasRightJoin = true;
+ }
+ }
+ if (statement.getWhere() != null) {
+ if (hasRightJoin) {
+ // conditions can't be pushed down to the scan filter.
+ postFilters.add(statement.getWhere());
+ } else {
+ statement.getWhere().accept(new WhereNodeVisitor(resolver));
+ for (ParseNode prefilter : preFilters) {
+ prefilter.accept(prefilterRefVisitor);
+ }
+ }
+ for (ParseNode postfilter : postFilters) {
+ postfilter.accept(generalRefVisitor);
+ }
+ }
+ for (AliasedNode node : selectList) {
+ node.getNode().accept(generalRefVisitor);
+ }
+ if (statement.getGroupBy() != null) {
+ for (ParseNode node : statement.getGroupBy()) {
+ node.accept(generalRefVisitor);
+ }
+ }
+ if (statement.getHaving() != null) {
+ statement.getHaving().accept(generalRefVisitor);
+ }
+ if (statement.getOrderBy() != null) {
+ for (OrderByNode node : statement.getOrderBy()) {
+ node.getNode().accept(generalRefVisitor);
+ }
+ }
+ this.columnRefs = new HashMap<ColumnRef, ColumnRefType>();
+ for (ColumnRef ref : generalRefVisitor.getColumnRefMap().keySet()) {
+ columnRefs.put(ref, ColumnRefType.GENERAL);
+ }
+ for (ColumnRef ref : joinLocalRefVisitor.getColumnRefMap().keySet()) {
+ if (!columnRefs.containsKey(ref))
+ columnRefs.put(ref, ColumnRefType.JOINLOCAL);
+ }
+ for (ColumnRef ref : prefilterRefVisitor.getColumnRefMap().keySet()) {
+ if (!columnRefs.containsKey(ref))
+ columnRefs.put(ref, ColumnRefType.PREFILTER);
+ }
+ }
+
+ private JoinSpec(TableRef table, List<AliasedNode> select, List<ParseNode> preFilters,
+ List<ParseNode> postFilters, List<JoinTable> joinTables, Map<ColumnRef, ColumnRefType> columnRefs) {
+ this.mainTable = table;
+ this.select = select;
+ this.preFilters = preFilters;
+ this.postFilters = postFilters;
+ this.joinTables = joinTables;
+ this.columnRefs = columnRefs;
+ }
+
+ public TableRef getMainTable() {
+ return mainTable;
+ }
+
+ public List<AliasedNode> getSelect() {
+ return select;
+ }
+
+ public List<ParseNode> getPreFilters() {
+ return preFilters;
+ }
+
+ public List<ParseNode> getPostFilters() {
+ return postFilters;
+ }
+
+ public List<JoinTable> getJoinTables() {
+ return joinTables;
+ }
+
+ public ParseNode getPreFiltersCombined() {
+ if (preFilters == null || preFilters.isEmpty())
+ return null;
+
+ if (preFilters.size() == 1)
+ return preFilters.get(0);
+
+ return NODE_FACTORY.and(preFilters);
+ }
+
+ public Expression compilePostFilterExpression(StatementContext context) throws SQLException {
+ if (postFilters == null || postFilters.isEmpty())
+ return null;
+
+ ExpressionCompiler expressionCompiler = new ExpressionCompiler(context);
+ List<Expression> expressions = new ArrayList<Expression>(postFilters.size());
+ for (ParseNode postFilter : postFilters) {
+ expressionCompiler.reset();
+ Expression expression = postFilter.accept(expressionCompiler);
+ expressions.add(expression);
+ }
+
+ if (expressions.size() == 1)
+ return expressions.get(0);
+
+ return new AndExpression(expressions);
+ }
+
+ public void projectColumns(Scan scan, TableRef table) {
+ if (isWildCardSelect(select)) {
+ scan.getFamilyMap().clear();
+ return;
+ }
+ for (ColumnRef columnRef : columnRefs.keySet()) {
+ if (columnRef.getTableRef().equals(table)
+ && !SchemaUtil.isPKColumn(columnRef.getColumn())) {
+ scan.addColumn(columnRef.getColumn().getFamilyName().getBytes(), columnRef.getColumn().getName().getBytes());
+ }
+ }
+ }
+
+ public ProjectedPTableWrapper createProjectedTable(TableRef tableRef, boolean retainPKColumns) throws SQLException {
+ List<PColumn> projectedColumns = new ArrayList<PColumn>();
+ List<Expression> sourceExpressions = new ArrayList<Expression>();
+ ListMultimap<String, String> columnNameMap = ArrayListMultimap.<String, String>create();
+ PTable table = tableRef.getTable();
+ boolean hasSaltingColumn = retainPKColumns && table.getBucketNum() != null;
+ if (retainPKColumns) {
+ for (PColumn column : table.getPKColumns()) {
+ addProjectedColumn(projectedColumns, sourceExpressions, columnNameMap,
+ column, tableRef, column.getFamilyName(), hasSaltingColumn);
+ }
+ }
+ if (isWildCardSelect(select)) {
+ for (PColumn column : table.getColumns()) {
+ if (!retainPKColumns || !SchemaUtil.isPKColumn(column)) {
+ addProjectedColumn(projectedColumns, sourceExpressions, columnNameMap,
+ column, tableRef, PNameFactory.newName(ScanProjector.VALUE_COLUMN_FAMILY), hasSaltingColumn);
+ }
+ }
+ } else {
+ for (Map.Entry<ColumnRef, ColumnRefType> e : columnRefs.entrySet()) {
+ ColumnRef columnRef = e.getKey();
+ if (e.getValue() != ColumnRefType.PREFILTER
+ && columnRef.getTableRef().equals(tableRef)
+ && (!retainPKColumns || !SchemaUtil.isPKColumn(columnRef.getColumn()))) {
+ PColumn column = columnRef.getColumn();
+ addProjectedColumn(projectedColumns, sourceExpressions, columnNameMap,
+ column, tableRef, PNameFactory.newName(ScanProjector.VALUE_COLUMN_FAMILY), hasSaltingColumn);
+ }
+ }
+ }
+
+ PTable t = PTableImpl.makePTable(PNameFactory.newName(PROJECTED_TABLE_SCHEMA), table.getName(), PTableType.JOIN, table.getIndexState(),
+ table.getTimeStamp(), table.getSequenceNumber(), table.getPKName(), retainPKColumns ? table.getBucketNum() : null,
+ projectedColumns, table.getParentTableName(), table.getIndexes(),
+ table.isImmutableRows(), Collections.<PName>emptyList(), null, null, table.isWALDisabled(), table.isMultiTenant(), table.getViewType());
+ return new ProjectedPTableWrapper(t, columnNameMap, sourceExpressions);
+ }
+
+ private static void addProjectedColumn(List<PColumn> projectedColumns, List<Expression> sourceExpressions,
+ ListMultimap<String, String> columnNameMap, PColumn sourceColumn, TableRef sourceTable, PName familyName, boolean hasSaltingColumn)
+ throws SQLException {
+ if (sourceColumn == SALTING_COLUMN)
+ return;
+
+ int position = projectedColumns.size() + (hasSaltingColumn ? 1 : 0);
+ PTable table = sourceTable.getTable();
+ PName colName = sourceColumn.getName();
+ PName name = sourceTable.getTableAlias() == null ? null : PNameFactory.newName(getProjectedColumnName(null, sourceTable.getTableAlias(), colName.getString()));
+ PName fullName = getProjectedColumnName(table.getSchemaName(), table.getTableName(), colName);
+ if (name == null) {
+ name = fullName;
+ } else {
+ columnNameMap.put(fullName.getString(), name.getString());
+ }
+ columnNameMap.put(colName.getString(), name.getString());
+ PColumnImpl column = new PColumnImpl(name, familyName, sourceColumn.getDataType(),
+ sourceColumn.getMaxLength(), sourceColumn.getScale(), sourceColumn.isNullable(),
+ position, sourceColumn.getColumnModifier(), sourceColumn.getArraySize());
+ Expression sourceExpression = new ColumnRef(sourceTable, sourceColumn.getPosition()).newColumnExpression();
+ projectedColumns.add(column);
+ sourceExpressions.add(sourceExpression);
+ }
+
+ public boolean hasPostReference(TableRef table) {
+ if (isWildCardSelect(select))
+ return true;
+
+ for (Map.Entry<ColumnRef, ColumnRefType> e : columnRefs.entrySet()) {
+ if (e.getValue() == ColumnRefType.GENERAL && e.getKey().getTableRef().equals(table)) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ private class WhereNodeVisitor extends TraverseNoParseNodeVisitor<Void> {
+ private ColumnResolver resolver;
+
+ public WhereNodeVisitor(ColumnResolver resolver) {
+ this.resolver = resolver;
+ }
+
+ private Void leaveBooleanNode(ParseNode node,
+ List<Void> l) throws SQLException {
+ ColumnParseNodeVisitor visitor = new ColumnParseNodeVisitor(resolver);
+ node.accept(visitor);
+ ColumnParseNodeVisitor.ContentType type = visitor.getContentType(mainTable);
+ if (type == ColumnParseNodeVisitor.ContentType.NONE
+ || type == ColumnParseNodeVisitor.ContentType.SELF_ONLY) {
+ preFilters.add(node);
+ } else {
+ postFilters.add(node);
+ }
+ return null;
+ }
+
+ @Override
+ public Void visitLeave(LikeParseNode node,
+ List<Void> l) throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public boolean visitEnter(AndParseNode node) {
+ return true;
+ }
+
+ @Override
+ public Void visitLeave(OrParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(ComparisonParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(NotParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(InListParseNode node,
+ List<Void> l) throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(IsNullParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(FunctionParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(BetweenParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(CaseParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(CastParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveBooleanNode(node, l);
+ }
+ }
+ }
+
+ public static JoinSpec getSubJoinSpecWithoutPostFilters(JoinSpec join) {
+ return new JoinSpec(join.mainTable, join.select, join.preFilters, new ArrayList<ParseNode>(),
+ join.joinTables.subList(0, join.joinTables.size() - 1), join.columnRefs);
+ }
+
+ public static class JoinTable {
+ private JoinType type;
+ private TableNode tableNode; // original table node
+ private TableRef table;
+ private List<AliasedNode> select; // all basic nodes related to this table, no aggregation.
+ private List<ParseNode> preFilters;
+ private List<ParseNode> conditions;
+ private SelectStatement subquery;
+
+ private Set<TableRef> leftTableRefs;
+
+ public JoinTable(JoinTableNode node, TableRef tableRef, List<AliasedNode> select, ColumnResolver resolver) throws SQLException {
+ if (!(node.getTable() instanceof ConcreteTableNode))
+ throw new SQLFeatureNotSupportedException("Subqueries not supported.");
+
+ this.type = node.getType();
+ this.tableNode = node.getTable();
+ this.table = tableRef;
+ this.select = extractFromSelect(select,tableRef,resolver);
+ this.preFilters = new ArrayList<ParseNode>();
+ this.conditions = new ArrayList<ParseNode>();
+ this.leftTableRefs = new HashSet<TableRef>();
+ node.getOnNode().accept(new OnNodeVisitor(resolver));
+ }
+
+ public JoinType getType() {
+ return type;
+ }
+
+ public TableNode getTableNode() {
+ return tableNode;
+ }
+
+ public TableRef getTable() {
+ return table;
+ }
+
+ public List<AliasedNode> getSelect() {
+ return select;
+ }
+
+ public List<ParseNode> getPreFilters() {
+ return preFilters;
+ }
+
+ public List<ParseNode> getJoinConditions() {
+ return conditions;
+ }
+
+ public SelectStatement getSubquery() {
+ return subquery;
+ }
+
+ public Set<TableRef> getLeftTableRefs() {
+ return leftTableRefs;
+ }
+
+ public ParseNode getPreFiltersCombined() {
+ if (preFilters == null || preFilters.isEmpty())
+ return null;
+
+ if (preFilters.size() == 1)
+ return preFilters.get(0);
+
+ return NODE_FACTORY.and(preFilters);
+ }
+
+ public SelectStatement getAsSubquery() {
+ if (subquery != null)
+ return subquery;
+
+ List<TableNode> from = new ArrayList<TableNode>(1);
+ from.add(tableNode);
+ return NODE_FACTORY.select(from, null, false, select, getPreFiltersCombined(), null, null, null, null, 0, false);
+ }
+
+ public Pair<List<Expression>, List<Expression>> compileJoinConditions(StatementContext context, ColumnResolver leftResolver, ColumnResolver rightResolver) throws SQLException {
+ ColumnResolver resolver = context.getResolver();
+ List<Pair<Expression, Expression>> compiled = new ArrayList<Pair<Expression, Expression>>(conditions.size());
+ context.setResolver(leftResolver);
+ ExpressionCompiler expressionCompiler = new ExpressionCompiler(context);
+ for (ParseNode condition : conditions) {
+ assert (condition instanceof EqualParseNode);
+ EqualParseNode equalNode = (EqualParseNode) condition;
+ expressionCompiler.reset();
+ Expression left = equalNode.getLHS().accept(expressionCompiler);
+ compiled.add(new Pair<Expression, Expression>(left, null));
+ }
+ context.setResolver(rightResolver);
+ expressionCompiler = new ExpressionCompiler(context);
+ Iterator<Pair<Expression, Expression>> iter = compiled.iterator();
+ for (ParseNode condition : conditions) {
+ Pair<Expression, Expression> p = iter.next();
+ EqualParseNode equalNode = (EqualParseNode) condition;
+ expressionCompiler.reset();
+ Expression right = equalNode.getRHS().accept(expressionCompiler);
+ Expression left = p.getFirst();
+ PDataType toType = getCommonType(left.getDataType(), right.getDataType());
+ if (left.getDataType() != toType) {
+ left = CoerceExpression.create(left, toType);
+ p.setFirst(left);
+ }
+ if (right.getDataType() != toType) {
+ right = CoerceExpression.create(right, toType);
+ }
+ p.setSecond(right);
+ }
+ context.setResolver(resolver); // recover the resolver
+ Collections.sort(compiled, new Comparator<Pair<Expression, Expression>>() {
+ @Override
+ public int compare(Pair<Expression, Expression> o1, Pair<Expression, Expression> o2) {
+ Expression e1 = o1.getFirst();
+ Expression e2 = o2.getFirst();
+ boolean isFixed1 = e1.getDataType().isFixedWidth();
+ boolean isFixed2 = e2.getDataType().isFixedWidth();
+ boolean isFixedNullable1 = e1.isNullable() &&isFixed1;
+ boolean isFixedNullable2 = e2.isNullable() && isFixed2;
+ if (isFixedNullable1 == isFixedNullable2) {
+ if (isFixed1 == isFixed2) {
+ return 0;
+ } else if (isFixed1) {
+ return -1;
+ } else {
+ return 1;
+ }
+ } else if (isFixedNullable1) {
+ return 1;
+ } else {
+ return -1;
+ }
+ }
+ });
+ List<Expression> lConditions = new ArrayList<Expression>(compiled.size());
+ List<Expression> rConditions = new ArrayList<Expression>(compiled.size());
+ for (Pair<Expression, Expression> pair : compiled) {
+ lConditions.add(pair.getFirst());
+ rConditions.add(pair.getSecond());
+ }
+
+ return new Pair<List<Expression>, List<Expression>>(lConditions, rConditions);
+ }
+
+ private PDataType getCommonType(PDataType lType, PDataType rType) throws SQLException {
+ if (lType == rType)
+ return lType;
+
+ if (!lType.isComparableTo(rType))
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.CANNOT_CONVERT_TYPE)
+ .setMessage("On-clause LHS expression and RHS expression must be comparable. LHS type: " + lType + ", RHS type: " + rType)
+ .build().buildException();
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.TINYINT))
+ && (rType == null || rType.isCoercibleTo(PDataType.TINYINT))) {
+ return lType == null ? rType : lType; // to preserve UNSIGNED type
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.SMALLINT))
+ && (rType == null || rType.isCoercibleTo(PDataType.SMALLINT))) {
+ return lType == null ? rType : lType; // to preserve UNSIGNED type
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.INTEGER))
+ && (rType == null || rType.isCoercibleTo(PDataType.INTEGER))) {
+ return lType == null ? rType : lType; // to preserve UNSIGNED type
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.LONG))
+ && (rType == null || rType.isCoercibleTo(PDataType.LONG))) {
+ return lType == null ? rType : lType; // to preserve UNSIGNED type
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.DECIMAL))
+ && (rType == null || rType.isCoercibleTo(PDataType.DECIMAL))) {
+ return PDataType.DECIMAL;
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.DATE))
+ && (rType == null || rType.isCoercibleTo(PDataType.DATE))) {
+ return lType == null ? rType : lType;
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.TIMESTAMP))
+ && (rType == null || rType.isCoercibleTo(PDataType.TIMESTAMP))) {
+ return lType == null ? rType : lType;
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.VARCHAR))
+ && (rType == null || rType.isCoercibleTo(PDataType.VARCHAR))) {
+ return PDataType.VARCHAR;
+ }
+
+ if ((lType == null || lType.isCoercibleTo(PDataType.BOOLEAN))
+ && (rType == null || rType.isCoercibleTo(PDataType.BOOLEAN))) {
+ return PDataType.BOOLEAN;
+ }
+
+ return PDataType.VARBINARY;
+ }
+
+ private class OnNodeVisitor extends TraverseNoParseNodeVisitor<Void> {
+ private ColumnResolver resolver;
+
+ public OnNodeVisitor(ColumnResolver resolver) {
+ this.resolver = resolver;
+ }
+
+ private Void leaveNonEqBooleanNode(ParseNode node,
+ List<Void> l) throws SQLException {
+ ColumnParseNodeVisitor visitor = new ColumnParseNodeVisitor(resolver);
+ node.accept(visitor);
+ ColumnParseNodeVisitor.ContentType type = visitor.getContentType(table);
+ if (type == ColumnParseNodeVisitor.ContentType.NONE
+ || type == ColumnParseNodeVisitor.ContentType.SELF_ONLY) {
+ preFilters.add(node);
+ } else {
+ throwUnsupportedJoinConditionException();
+ }
+ return null;
+ }
+
+ @Override
+ public Void visitLeave(LikeParseNode node,
+ List<Void> l) throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public boolean visitEnter(AndParseNode node) {
+ return true;
+ }
+
+ @Override
+ public Void visitLeave(OrParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(ComparisonParseNode node, List<Void> l)
+ throws SQLException {
+ if (!(node instanceof EqualParseNode))
+ return leaveNonEqBooleanNode(node, l);
+ ColumnParseNodeVisitor lhsVisitor = new ColumnParseNodeVisitor(resolver);
+ ColumnParseNodeVisitor rhsVisitor = new ColumnParseNodeVisitor(resolver);
+ node.getLHS().accept(lhsVisitor);
+ node.getRHS().accept(rhsVisitor);
+ ColumnParseNodeVisitor.ContentType lhsType = lhsVisitor.getContentType(table);
+ ColumnParseNodeVisitor.ContentType rhsType = rhsVisitor.getContentType(table);
+ if ((lhsType == ColumnParseNodeVisitor.ContentType.SELF_ONLY || lhsType == ColumnParseNodeVisitor.ContentType.NONE)
+ && (rhsType == ColumnParseNodeVisitor.ContentType.SELF_ONLY || rhsType == ColumnParseNodeVisitor.ContentType.NONE)) {
+ preFilters.add(node);
+ } else if (lhsType == ColumnParseNodeVisitor.ContentType.FOREIGN_ONLY
+ && rhsType == ColumnParseNodeVisitor.ContentType.SELF_ONLY) {
+ conditions.add(node);
+ leftTableRefs.addAll(lhsVisitor.getTableRefSet());
+ } else if (rhsType == ColumnParseNodeVisitor.ContentType.FOREIGN_ONLY
+ && lhsType == ColumnParseNodeVisitor.ContentType.SELF_ONLY) {
+ conditions.add(NODE_FACTORY.equal(node.getRHS(), node.getLHS()));
+ leftTableRefs.addAll(rhsVisitor.getTableRefSet());
+ } else {
+ throwUnsupportedJoinConditionException();
+ }
+ return null;
+ }
+
+ @Override
+ public Void visitLeave(NotParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(InListParseNode node,
+ List<Void> l) throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(IsNullParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(FunctionParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(BetweenParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(CaseParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ @Override
+ public Void visitLeave(CastParseNode node, List<Void> l)
+ throws SQLException {
+ return leaveNonEqBooleanNode(node, l);
+ }
+
+ /*
+ * Conditions in the ON clause can only be:
+ * 1) an equal test between a self table expression and a foreign
+ * table expression.
+ * 2) a boolean condition referencing to the self table only.
+ * Otherwise, it can be ambiguous.
+ */
+ private void throwUnsupportedJoinConditionException()
+ throws SQLFeatureNotSupportedException {
+ throw new SQLFeatureNotSupportedException("Does not support non-standard or non-equi join conditions.");
+ }
+ }
+ }
+
+ private static class ColumnParseNodeVisitor extends StatelessTraverseAllParseNodeVisitor {
+ public enum ContentType {NONE, SELF_ONLY, FOREIGN_ONLY, COMPLEX};
+
+ private ColumnResolver resolver;
+ private final Set<TableRef> tableRefSet;
+ private final Map<ColumnRef, ColumnParseNode> columnRefMap;
+
+ public ColumnParseNodeVisitor(ColumnResolver resolver) {
+ this.resolver = resolver;
+ this.tableRefSet = new HashSet<TableRef>();
+ this.columnRefMap = new HashMap<ColumnRef, ColumnParseNode>();
+ }
+
+ public void reset() {
+ this.tableRefSet.clear();
+ this.columnRefMap.clear();
+ }
+
+ @Override
+ public Void visit(ColumnParseNode node) throws SQLException {
+ ColumnRef columnRef = resolver.resolveColumn(node.getSchemaName(), node.getTableName(), node.getName());
+ columnRefMap.put(columnRef, node);
+ tableRefSet.add(columnRef.getTableRef());
+ return null;
+ }
+
+ public Set<TableRef> getTableRefSet() {
+ return tableRefSet;
+ }
+
+ public Map<ColumnRef, ColumnParseNode> getColumnRefMap() {
+ return columnRefMap;
+ }
+
+ public ContentType getContentType(TableRef selfTable) {
+ if (tableRefSet.isEmpty())
+ return ContentType.NONE;
+ if (tableRefSet.size() > 1)
+ return ContentType.COMPLEX;
+ if (tableRefSet.contains(selfTable))
+ return ContentType.SELF_ONLY;
+ return ContentType.FOREIGN_ONLY;
+ }
+ }
+
+ private static String PROJECTED_TABLE_SCHEMA = ".";
+ // for creation of new statements
+ private static ParseNodeFactory NODE_FACTORY = new ParseNodeFactory();
+
+ private static boolean isWildCardSelect(List<AliasedNode> select) {
+ return (select.size() == 1 && select.get(0).getNode() == WildcardParseNode.INSTANCE);
+ }
+
+ private static List<AliasedNode> extractFromSelect(List<AliasedNode> select, TableRef table, ColumnResolver resolver) throws SQLException {
+ List<AliasedNode> ret = new ArrayList<AliasedNode>();
+ if (isWildCardSelect(select)) {
+ ret.add(NODE_FACTORY.aliasedNode(null, WildcardParseNode.INSTANCE));
+ return ret;
+ }
+
+ ColumnParseNodeVisitor visitor = new ColumnParseNodeVisitor(resolver);
+ for (AliasedNode node : select) {
+ node.getNode().accept(visitor);
+ ColumnParseNodeVisitor.ContentType type = visitor.getContentType(table);
+ if (type == ColumnParseNodeVisitor.ContentType.SELF_ONLY) {
+ ret.add(node);
+ } else if (type == ColumnParseNodeVisitor.ContentType.COMPLEX) {
+ for (Map.Entry<ColumnRef, ColumnParseNode> entry : visitor.getColumnRefMap().entrySet()) {
+ if (entry.getKey().getTableRef().equals(table)) {
+ ret.add(NODE_FACTORY.aliasedNode(null, entry.getValue()));
+ }
+ }
+ }
+ visitor.reset();
+ }
+ return ret;
+ }
+
+ public static JoinSpec getJoinSpec(StatementContext context, SelectStatement statement) throws SQLException {
+ return new JoinSpec(statement, context.getResolver());
+ }
+
+ public static SelectStatement optimize(StatementContext context, SelectStatement select, PhoenixStatement statement) throws SQLException {
+ ColumnResolver resolver = context.getResolver();
+ JoinSpec join = new JoinSpec(select, resolver);
+ Map<TableRef, TableRef> replacement = new HashMap<TableRef, TableRef>();
+ List<TableNode> from = select.getFrom();
+ List<TableNode> newFrom = Lists.newArrayListWithExpectedSize(from.size());
+
+ class TableNodeRewriter implements TableNodeVisitor {
+ private TableRef table;
+ private TableNode replaced;
+
+ TableNodeRewriter(TableRef table) {
+ this.table = table;
+ }
+
+ public TableNode getReplacedTableNode() {
+ return replaced;
+ }
+
+ @Override
+ public void visit(BindTableNode boundTableNode) throws SQLException {
+ replaced = NODE_FACTORY.bindTable(boundTableNode.getAlias(), getReplacedTableName());
+ }
+
+ @Override
+ public void visit(JoinTableNode joinNode) throws SQLException {
+ joinNode.getTable().accept(this);
+ replaced = NODE_FACTORY.join(joinNode.getType(), joinNode.getOnNode(), replaced);
+ }
+
+ @Override
+ public void visit(NamedTableNode namedTableNode)
+ throws SQLException {
+ replaced = NODE_FACTORY.namedTable(namedTableNode.getAlias(), getReplacedTableName(), namedTableNode.getDynamicColumns());
+ }
+
+ @Override
+ public void visit(DerivedTableNode subselectNode)
+ throws SQLException {
+ throw new SQLFeatureNotSupportedException();
+ }
+
+ private TableName getReplacedTableName() {
+ String schemaName = table.getTable().getSchemaName().getString();
+ schemaName = schemaName.length() == 0 ? null : '"' + schemaName + '"';
+ String tableName = '"' + table.getTable().getTableName().getString() + '"';
+ return NODE_FACTORY.table(schemaName, tableName);
+ }
+ };
+
+ // get optimized plans for join tables
+ for (int i = 1; i < from.size(); i++) {
+ TableNode jNode = from.get(i);
+ assert (jNode instanceof JoinTableNode);
+ TableNode tNode = ((JoinTableNode) jNode).getTable();
+ for (JoinTable jTable : join.getJoinTables()) {
+ if (jTable.getTableNode() != tNode)
+ continue;
+ TableRef table = jTable.getTable();
+ SelectStatement stmt = getSubqueryForOptimizedPlan(select, table, join.columnRefs, jTable.getPreFiltersCombined());
+ QueryPlan plan = context.getConnection().getQueryServices().getOptimizer().optimize(stmt, statement);
+ if (!plan.getTableRef().equals(table)) {
+ TableNodeRewriter rewriter = new TableNodeRewriter(plan.getTableRef());
+ jNode.accept(rewriter);
+ newFrom.add(rewriter.getReplacedTableNode());
+ replacement.put(table, plan.getTableRef());
+ } else {
+ newFrom.add(jNode);
+ }
+ }
+ }
+ // get optimized plan for main table
+ TableRef table = join.getMainTable();
+ SelectStatement stmt = getSubqueryForOptimizedPlan(select, table, join.columnRefs, join.getPreFiltersCombined());
+ QueryPlan plan = context.getConnection().getQueryServices().getOptimizer().optimize(stmt, statement);
+ if (!plan.getTableRef().equals(table)) {
+ TableNodeRewriter rewriter = new TableNodeRewriter(plan.getTableRef());
+ from.get(0).accept(rewriter);
+ newFrom.add(0, rewriter.getReplacedTableNode());
+ replacement.put(table, plan.getTableRef());
+ } else {
+ newFrom.add(0, from.get(0));
+ }
+
+ if (replacement.isEmpty())
+ return select;
+
+ return IndexStatementRewriter.translate(NODE_FACTORY.select(select, newFrom), resolver, replacement);
+ }
+
+ private static SelectStatement getSubqueryForOptimizedPlan(SelectStatement select, TableRef table, Map<ColumnRef, ColumnRefType> columnRefs, ParseNode where) {
+ TableName tName = NODE_FACTORY.table(table.getTable().getSchemaName().getString(), table.getTable().getTableName().getString());
+ List<AliasedNode> selectList = new ArrayList<AliasedNode>();
+ if (isWildCardSelect(select.getSelect())) {
+ selectList.add(NODE_FACTORY.aliasedNode(null, WildcardParseNode.INSTANCE));
+ } else {
+ for (ColumnRef colRef : columnRefs.keySet()) {
+ if (colRef.getTableRef().equals(table)) {
+ selectList.add(NODE_FACTORY.aliasedNode(null, NODE_FACTORY.column(tName, '"' + colRef.getColumn().getName().getString() + '"', null)));
+ }
+ }
+ }
+ List<? extends TableNode> from = Collections.singletonList(NODE_FACTORY.namedTable(table.getTableAlias(), tName));
+
+ return NODE_FACTORY.select(from, select.getHint(), false, selectList, where, null, null, null, null, 0, false);
+ }
+
+ /**
+ * Returns a boolean vector indicating whether the evaluation of join expressions
+ * can be evaluated at an early stage if the input JoinSpec can be taken as a
+ * star join. Otherwise returns null.
+ * @param join the JoinSpec
+ * @return a boolean vector for a star join; or null for non star join.
+ */
+ public static boolean[] getStarJoinVector(JoinSpec join) {
+ assert(!join.getJoinTables().isEmpty());
+
+ int count = join.getJoinTables().size();
+ boolean[] vector = new boolean[count];
+ for (int i = 0; i < count; i++) {
+ JoinTable joinTable = join.getJoinTables().get(i);
+ if (joinTable.getType() != JoinType.Left
+ && joinTable.getType() != JoinType.Inner)
+ return null;
+ vector[i] = true;
+ Iterator<TableRef> iter = joinTable.getLeftTableRefs().iterator();
+ while (vector[i] == true && iter.hasNext()) {
+ TableRef tableRef = iter.next();
+ if (!tableRef.equals(join.getMainTable())) {
+ vector[i] = false;
+ }
+ }
+ }
+
+ return vector;
+ }
+
+ public static SelectStatement getSubqueryWithoutJoin(SelectStatement statement, JoinSpec join) {
+ return NODE_FACTORY.select(statement.getFrom().subList(0, 1), statement.getHint(), statement.isDistinct(), statement.getSelect(), join.getPreFiltersCombined(), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ }
+
+ // Get the last join table select statement with fixed-up select and where nodes.
+ // Currently does NOT support last join table as a subquery.
+ public static SelectStatement getSubqueryForLastJoinTable(SelectStatement statement, JoinSpec join) throws SQLException {
+ List<JoinTable> joinTables = join.getJoinTables();
+ int count = joinTables.size();
+ assert (count > 0);
+ JoinTable lastJoinTable = joinTables.get(count - 1);
+ if (lastJoinTable.getSubquery() != null) {
+ throw new SQLFeatureNotSupportedException("Subqueries not supported.");
+ }
+ List<TableNode> from = new ArrayList<TableNode>(1);
+ from.add(lastJoinTable.getTableNode());
+
+ return NODE_FACTORY.select(from, statement.getHint(), statement.isDistinct(), statement.getSelect(), lastJoinTable.getPreFiltersCombined(), statement.getGroupBy(), statement.getHaving(), statement.getOrderBy(), statement.getLimit(), statement.getBindCount(), statement.isAggregate());
+ }
+
+ // Get subquery with fixed select and where nodes
+ public static SelectStatement getSubQueryWithoutLastJoin(SelectStatement statement, JoinSpec join) {
+ List<TableNode> from = statement.getFrom();
+ assert(from.size() > 1);
+ List<JoinTable> joinTables = join.getJoinTables();
+ int count = joinTables.size();
+ assert (count > 0);
+ List<AliasedNode> select = new ArrayList<AliasedNode>();
+ select.addAll(join.getSelect());
+ for (int i = 0; i < count - 1; i++) {
+ select.addAll(joinTables.get(i).getSelect());
+ }
+
+ return NODE_FACTORY.select(from.subList(0, from.size() - 1), statement.getHint(), false, select, join.getPreFiltersCombined(), null, null, null, null, statement.getBindCount(), false);
+ }
+
+ public static PTableWrapper mergeProjectedTables(PTableWrapper lWrapper, PTableWrapper rWrapper, boolean innerJoin) throws SQLException {
+ PTable left = lWrapper.getTable();
+ PTable right = rWrapper.getTable();
+ List<PColumn> merged = new ArrayList<PColumn>();
+ merged.addAll(left.getColumns());
+ int position = merged.size();
+ for (PColumn c : right.getColumns()) {
+ if (!SchemaUtil.isPKColumn(c)) {
+ PColumnImpl column = new PColumnImpl(c.getName(),
+ PNameFactory.newName(ScanProjector.VALUE_COLUMN_FAMILY), c.getDataType(),
+ c.getMaxLength(), c.getScale(), innerJoin ? c.isNullable() : true, position++,
+ c.getColumnModifier(), c.getArraySize());
+ merged.add(column);
+ }
+ }
+ if (left.getBucketNum() != null) {
+ merged.remove(0);
+ }
+ PTable t = PTableImpl.makePTable(left.getSchemaName(), PNameFactory.newName(SchemaUtil.getTableName(left.getName().getString(), right.getName().getString())),
+ left.getType(), left.getIndexState(), left.getTimeStamp(), left.getSequenceNumber(), left.getPKName(), left.getBucketNum(), merged, left.getParentTableName(),
+ left.getIndexes(), left.isImmutableRows(), Collections.<PName>emptyList(), null, null, PTable.DEFAULT_DISABLE_WAL, left.isMultiTenant(), left.getViewType());
+
+ ListMultimap<String, String> mergedMap = ArrayListMultimap.<String, String>create();
+ mergedMap.putAll(lWrapper.getColumnNameMap());
+ mergedMap.putAll(rWrapper.getColumnNameMap());
+
+ return new PTableWrapper(t, mergedMap);
+ }
+
+ public static ScanProjector getScanProjector(ProjectedPTableWrapper table) {
+ return new ScanProjector(table);
+ }
+
+ public static class PTableWrapper {
+ protected PTable table;
+ protected ListMultimap<String, String> columnNameMap;
+
+ protected PTableWrapper(PTable table, ListMultimap<String, String> columnNameMap) {
+ this.table = table;
+ this.columnNameMap = columnNameMap;
+ }
+
+ public PTable getTable() {
+ return table;
+ }
+
+ public ListMultimap<String, String> getColumnNameMap() {
+ return columnNameMap;
+ }
+
+ public List<String> getMappedColumnName(String name) {
+ return columnNameMap.get(name);
+ }
+ }
+
+ public static class ProjectedPTableWrapper extends PTableWrapper {
+ private List<Expression> sourceExpressions;
+
+ protected ProjectedPTableWrapper(PTable table, ListMultimap<String, String> columnNameMap, List<Expression> sourceExpressions) {
+ super(table, columnNameMap);
+ this.sourceExpressions = sourceExpressions;
+ }
+
+ public Expression getSourceExpression(PColumn column) {
+ return sourceExpressions.get(column.getPosition() - (table.getBucketNum() == null ? 0 : 1));
+ }
+ }
+
+ public static ColumnResolver getColumnResolver(PTableWrapper table) {
+ return new JoinedTableColumnResolver(table);
+ }
+
+ public static class JoinedTableColumnResolver implements ColumnResolver {
+ private PTableWrapper table;
+ private List<TableRef> tableRefs;
+
+ private JoinedTableColumnResolver(PTableWrapper table) {
+ this.table = table;
+ TableRef tableRef = new TableRef(null, table.getTable(), 0, false);
+ this.tableRefs = ImmutableList.of(tableRef);
+ }
+
+ @Override
+ public List<TableRef> getTables() {
+ return tableRefs;
+ }
+
+ public PTableWrapper getPTableWrapper() {
+ return table;
+ }
+
+ @Override
+ public ColumnRef resolveColumn(String schemaName, String tableName,
+ String colName) throws SQLException {
+ String name = getProjectedColumnName(schemaName, tableName, colName);
+ TableRef tableRef = tableRefs.get(0);
+ try {
+ PColumn column = tableRef.getTable().getColumn(name);
+ return new ColumnRef(tableRef, column.getPosition());
+ } catch (ColumnNotFoundException e) {
+ List<String> names = table.getMappedColumnName(name);
+ if (names.size() == 1) {
+ PColumn column = tableRef.getTable().getColumn(names.get(0));
+ return new ColumnRef(tableRef, column.getPosition());
+ }
+
+ if (names.size() > 1) {
+ throw new AmbiguousColumnException(name);
+ }
+
+ throw e;
+ }
+ }
+ }
+
+ private static String getProjectedColumnName(String schemaName, String tableName,
+ String colName) {
+ return SchemaUtil.getColumnName(SchemaUtil.getTableName(schemaName, tableName), colName);
+ }
+
+ private static PName getProjectedColumnName(PName schemaName, PName tableName,
+ PName colName) {
+ String name = getProjectedColumnName(schemaName.getString(), tableName.getString(), colName.getString());
+ return PNameFactory.newName(name);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/KeyPart.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/KeyPart.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/KeyPart.java
new file mode 100644
index 0000000..1c8b0eb
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/KeyPart.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+import java.util.List;
+
+import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
+
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.query.KeyRange;
+import org.apache.phoenix.schema.PColumn;
+
+/**
+ *
+ * Interface that determines how a key part contributes to
+ * the forming of the key (start/stop of scan and SkipScanFilter)
+ * for each part of a multi-part primary key. It acts as the glue
+ * between a built-in function and the setting of the scan key
+ * during query compilation.
+ *
+ * @author jtaylor
+ * @since 0.12
+ */
+public interface KeyPart {
+ /**
+ * Calculate the key range given an operator and the key on
+ * the RHS of an expression. For example, given the expression
+ * SUBSTR(foo,1,3) = 'bar', the key range would be ['bar','bas'),
+ * and if foo was fixed length, the upper and lower key range
+ * bytes would be filled out to the fixed length.
+ * @param op comparison operator (=, <=, <, >=, >, !=)
+ * @param rhs the constant on the RHS of an expression.
+ * @return the key range that encompasses the range for the
+ * expression for which this keyPart is associated
+ *
+ * @see org.apache.phoenix.expression.function.ScalarFunction#newKeyPart(KeyPart)
+ */
+ public KeyRange getKeyRange(CompareOp op, Expression rhs);
+
+ /**
+ * Determines whether an expression gets extracted from the
+ * WHERE clause if it contributes toward the building of the
+ * scan key. For example, the SUBSTR built-in function may
+ * be extracted, since it may be completely represented
+ * through a key range. However, the REGEXP_SUBSTR must be
+ * left in the WHERE clause, since only the constant prefix
+ * part of the evaluation can be represented through a key
+ * range (i.e. rows may pass through that will fail when
+ * the REGEXP_SUBSTR is evaluated).
+ *
+ * @return an empty list if the expression should remain in
+ * the WHEERE clause for post filtering or a singleton list
+ * containing the expression if it should be removed.
+ */
+ public List<Expression> getExtractNodes();
+
+ /**
+ * Gets the primary key column associated with the start of this key part.
+ * @return the primary key column for this key part
+ */
+ public PColumn getColumn();
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/LimitCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/LimitCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/LimitCompiler.java
new file mode 100644
index 0000000..5c4cbb2
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/LimitCompiler.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.parse.*;
+import org.apache.phoenix.schema.*;
+
+
+public class LimitCompiler {
+ private static final ParseNodeFactory NODE_FACTORY = new ParseNodeFactory();
+
+ public static final PDatum LIMIT_DATUM = new PDatum() {
+ @Override
+ public boolean isNullable() {
+ return false;
+ }
+ @Override
+ public PDataType getDataType() {
+ return PDataType.INTEGER;
+ }
+ @Override
+ public Integer getByteSize() {
+ return getDataType().getByteSize();
+ }
+ @Override
+ public Integer getMaxLength() {
+ return null;
+ }
+ @Override
+ public Integer getScale() {
+ return null;
+ }
+ @Override
+ public ColumnModifier getColumnModifier() {
+ return null;
+ }
+ };
+
+ private LimitCompiler() {
+ }
+
+ public static Integer compile(StatementContext context, FilterableStatement statement) throws SQLException {
+ LimitNode limitNode = statement.getLimit();
+ if (limitNode == null) {
+ return null;
+ }
+ LimitParseNodeVisitor visitor = new LimitParseNodeVisitor(context);
+ limitNode.getLimitParseNode().accept(visitor);
+ return visitor.getLimit();
+ }
+
+ private static class LimitParseNodeVisitor extends TraverseNoParseNodeVisitor<Void> {
+ private final StatementContext context;
+ private Integer limit;
+
+ private LimitParseNodeVisitor(StatementContext context) {
+ this.context = context;
+ }
+
+ public Integer getLimit() {
+ return limit;
+ }
+
+ @Override
+ public Void visit(LiteralParseNode node) throws SQLException {
+ Object limitValue = node.getValue();
+ // If limit is null, leave this.limit set to zero
+ // This means that we've bound limit to null for the purpose of
+ // collecting parameter metadata.
+ if (limitValue != null) {
+ Integer limit = (Integer)LIMIT_DATUM.getDataType().toObject(limitValue, node.getType());
+ if (limit.intValue() >= 0) { // TODO: handle LIMIT 0
+ this.limit = limit;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public Void visit(BindParseNode node) throws SQLException {
+ Object value = context.getBindManager().getBindValue(node);
+ context.getBindManager().addParamMetaData(node, LIMIT_DATUM);
+ // Resolve the bind value, create a LiteralParseNode, and call the visit method for it.
+ // In this way, we can deal with just having a literal on one side of the expression.
+ visit(NODE_FACTORY.literal(value, LIMIT_DATUM.getDataType()));
+ return null;
+ }
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/MutatingParallelIteratorFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/MutatingParallelIteratorFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/MutatingParallelIteratorFactory.java
new file mode 100644
index 0000000..b19af43
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/MutatingParallelIteratorFactory.java
@@ -0,0 +1,115 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+import static org.apache.phoenix.query.QueryConstants.AGG_TIMESTAMP;
+import static org.apache.phoenix.query.QueryConstants.SINGLE_COLUMN;
+import static org.apache.phoenix.query.QueryConstants.SINGLE_COLUMN_FAMILY;
+import static org.apache.phoenix.query.QueryConstants.UNGROUPED_AGG_ROW_KEY;
+
+import java.sql.SQLException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.KeyValue;
+
+import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.iterate.ParallelIterators.ParallelIteratorFactory;
+import org.apache.phoenix.iterate.PeekingResultIterator;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.query.ConnectionQueryServices;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.tuple.SingleKeyValueTuple;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.util.KeyValueUtil;
+
+/**
+ * Factory class used to instantiate an iterator to handle mutations made during a parallel scan.
+ */
+public abstract class MutatingParallelIteratorFactory implements ParallelIteratorFactory {
+ protected final PhoenixConnection connection;
+ protected final TableRef tableRef;
+
+ protected MutatingParallelIteratorFactory(PhoenixConnection connection, TableRef tableRef) {
+ this.connection = connection;
+ this.tableRef = tableRef;
+ }
+
+ /**
+ * Method that does the actual mutation work
+ */
+ abstract protected MutationState mutate(PhoenixConnection connection, ResultIterator iterator) throws SQLException;
+
+ @Override
+ public PeekingResultIterator newIterator(ResultIterator iterator) throws SQLException {
+ // Clone the connection as it's not thread safe and will be operated on in parallel
+ final PhoenixConnection connection = new PhoenixConnection(this.connection);
+ MutationState state = mutate(connection, iterator);
+ long totalRowCount = state.getUpdateCount();
+ if (connection.getAutoCommit()) {
+ connection.getMutationState().join(state);
+ connection.commit();
+ ConnectionQueryServices services = connection.getQueryServices();
+ int maxSize = services.getProps().getInt(QueryServices.MAX_MUTATION_SIZE_ATTRIB,QueryServicesOptions.DEFAULT_MAX_MUTATION_SIZE);
+ state = new MutationState(maxSize, connection, totalRowCount);
+ }
+ final MutationState finalState = state;
+ byte[] value = PDataType.LONG.toBytes(totalRowCount);
+ KeyValue keyValue = KeyValueUtil.newKeyValue(UNGROUPED_AGG_ROW_KEY, SINGLE_COLUMN_FAMILY, SINGLE_COLUMN, AGG_TIMESTAMP, value, 0, value.length);
+ final Tuple tuple = new SingleKeyValueTuple(keyValue);
+ return new PeekingResultIterator() {
+ private boolean done = false;
+
+ @Override
+ public Tuple next() throws SQLException {
+ if (done) {
+ return null;
+ }
+ done = true;
+ return tuple;
+ }
+
+ @Override
+ public void explain(List<String> planSteps) {
+ }
+
+ @Override
+ public void close() throws SQLException {
+ try {
+ // Join the child mutation states in close, since this is called in a single threaded manner
+ // after the parallel results have been processed.
+ if (!connection.getAutoCommit()) {
+ MutatingParallelIteratorFactory.this.connection.getMutationState().join(finalState);
+ }
+ } finally {
+ connection.close();
+ }
+ }
+
+ @Override
+ public Tuple peek() throws SQLException {
+ return done ? null : tuple;
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/MutationPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/MutationPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/MutationPlan.java
new file mode 100644
index 0000000..11fd2cd
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/MutationPlan.java
@@ -0,0 +1,31 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+import java.sql.SQLException;
+
+import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+
+
+public interface MutationPlan extends StatementPlan {
+ public PhoenixConnection getConnection();
+ public MutationState execute() throws SQLException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/OrderByCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/OrderByCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/OrderByCompiler.java
new file mode 100644
index 0000000..61c0fcc
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/OrderByCompiler.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
+import org.apache.phoenix.compile.TrackOrderPreservingExpressionCompiler.Ordering;
+import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.exception.SQLExceptionInfo;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.OrderByExpression;
+import org.apache.phoenix.parse.FilterableStatement;
+import org.apache.phoenix.parse.OrderByNode;
+import org.apache.phoenix.query.ConnectionQueryServices.Feature;
+import org.apache.phoenix.schema.ColumnModifier;
+
+/**
+ * Validates ORDER BY clause and builds up a list of referenced columns.
+ *
+ * @author syyang
+ * @since 0.1
+ */
+public class OrderByCompiler {
+ public static class OrderBy {
+ public static final OrderBy EMPTY_ORDER_BY = new OrderBy(Collections.<OrderByExpression>emptyList());
+ /**
+ * Used to indicate that there was an ORDER BY, but it was optimized out because
+ * rows are already returned in this order.
+ */
+ public static final OrderBy FWD_ROW_KEY_ORDER_BY = new OrderBy(Collections.<OrderByExpression>emptyList());
+ public static final OrderBy REV_ROW_KEY_ORDER_BY = new OrderBy(Collections.<OrderByExpression>emptyList());
+
+ private final List<OrderByExpression> orderByExpressions;
+
+ private OrderBy(List<OrderByExpression> orderByExpressions) {
+ this.orderByExpressions = ImmutableList.copyOf(orderByExpressions);
+ }
+
+ public List<OrderByExpression> getOrderByExpressions() {
+ return orderByExpressions;
+ }
+ }
+ /**
+ * Gets a list of columns in the ORDER BY clause
+ * @param context the query context for tracking various states
+ * associated with the given select statement
+ * @param statement TODO
+ * @param groupBy the list of columns in the GROUP BY clause
+ * @param limit the row limit or null if no limit
+ * @return the compiled ORDER BY clause
+ * @throws SQLException
+ */
+ public static OrderBy compile(StatementContext context,
+ FilterableStatement statement,
+ GroupBy groupBy, Integer limit) throws SQLException {
+ List<OrderByNode> orderByNodes = statement.getOrderBy();
+ if (orderByNodes.isEmpty()) {
+ return OrderBy.EMPTY_ORDER_BY;
+ }
+ // accumulate columns in ORDER BY
+ TrackOrderPreservingExpressionCompiler visitor =
+ new TrackOrderPreservingExpressionCompiler(context, groupBy,
+ orderByNodes.size(), Ordering.ORDERED);
+ LinkedHashSet<OrderByExpression> orderByExpressions = Sets.newLinkedHashSetWithExpectedSize(orderByNodes.size());
+ for (OrderByNode node : orderByNodes) {
+ boolean isAscending = node.isAscending();
+ Expression expression = node.getNode().accept(visitor);
+ if (!expression.isStateless() && visitor.addEntry(expression, isAscending ? null : ColumnModifier.SORT_DESC)) {
+ // Detect mix of aggregate and non aggregates (i.e. ORDER BY txns, SUM(txns)
+ if (!visitor.isAggregate()) {
+ if (statement.isAggregate() || statement.isDistinct()) {
+ // Detect ORDER BY not in SELECT DISTINCT: SELECT DISTINCT count(*) FROM t ORDER BY x
+ if (statement.isDistinct()) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.ORDER_BY_NOT_IN_SELECT_DISTINCT)
+ .setMessage(expression.toString()).build().buildException();
+ }
+ ExpressionCompiler.throwNonAggExpressionInAggException(expression.toString());
+ }
+ }
+ if (expression.getDataType().isArrayType()) {
+ throw new SQLExceptionInfo.Builder(SQLExceptionCode.ORDER_BY_ARRAY_NOT_SUPPORTED)
+ .setMessage(expression.toString()).build().buildException();
+ }
+ if (expression.getColumnModifier() == ColumnModifier.SORT_DESC) {
+ isAscending = !isAscending;
+ }
+ OrderByExpression orderByExpression = new OrderByExpression(expression, node.isNullsLast(), isAscending);
+ orderByExpressions.add(orderByExpression);
+ }
+ visitor.reset();
+ }
+
+ if (orderByExpressions.isEmpty()) {
+ return OrderBy.EMPTY_ORDER_BY;
+ }
+ // If we're ordering by the order returned by the scan, we don't need an order by
+ if (visitor.isOrderPreserving()) {
+ if (visitor.isReverse()) {
+ if (context.getConnection().getQueryServices().supportsFeature(Feature.REVERSE_SCAN)) {
+ return OrderBy.REV_ROW_KEY_ORDER_BY;
+ }
+ } else {
+ return OrderBy.FWD_ROW_KEY_ORDER_BY;
+ }
+ }
+
+ return new OrderBy(Lists.newArrayList(orderByExpressions.iterator()));
+ }
+
+
+ private OrderByCompiler() {
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-phoenix/blob/50d523f6/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java
new file mode 100644
index 0000000..40fe1c3
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/PostDDLCompiler.java
@@ -0,0 +1,225 @@
+/*
+ * Copyright 2010 The Apache Software Foundation
+ *
+ * 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.compile;
+
+import java.sql.ParameterMetaData;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
+import org.apache.phoenix.compile.GroupByCompiler.GroupBy;
+import org.apache.phoenix.compile.OrderByCompiler.OrderBy;
+import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver;
+import org.apache.phoenix.execute.AggregatePlan;
+import org.apache.phoenix.execute.MutationState;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixParameterMetaData;
+import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.parse.SelectStatement;
+import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.ColumnRef;
+import org.apache.phoenix.schema.PColumn;
+import org.apache.phoenix.schema.PColumnFamily;
+import org.apache.phoenix.schema.PDataType;
+import org.apache.phoenix.schema.TableRef;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.util.ScanUtil;
+
+import com.google.common.collect.Lists;
+
+
+/**
+ *
+ * Class that compiles plan to update data values after a DDL command
+ * executes.
+ *
+ * TODO: get rid of this ugly code and just go through the standard APIs.
+ * The only time we may still need this is to manage updating the empty
+ * key value, as we sometimes need to "go back through time" to adjust
+ * this.
+ * @author jtaylor
+ * @since 0.1
+ */
+public class PostDDLCompiler {
+ private final PhoenixConnection connection;
+
+ public PostDDLCompiler(PhoenixConnection connection) {
+ this.connection = connection;
+ }
+
+ public MutationPlan compile(final List<TableRef> tableRefs, final byte[] emptyCF, final byte[] projectCF, final List<PColumn> deleteList,
+ final long timestamp) throws SQLException {
+
+ return new MutationPlan() {
+
+ @Override
+ public PhoenixConnection getConnection() {
+ return connection;
+ }
+
+ @Override
+ public ParameterMetaData getParameterMetaData() {
+ return PhoenixParameterMetaData.EMPTY_PARAMETER_META_DATA;
+ }
+
+ @Override
+ public ExplainPlan getExplainPlan() throws SQLException {
+ return ExplainPlan.EMPTY_PLAN;
+ }
+
+ @Override
+ public MutationState execute() throws SQLException {
+ if (tableRefs.isEmpty()) {
+ return null;
+ }
+ boolean wasAutoCommit = connection.getAutoCommit();
+ try {
+ connection.setAutoCommit(true);
+ SQLException sqlE = null;
+ if (deleteList == null && emptyCF == null) {
+ return new MutationState(0, connection);
+ }
+ /*
+ * Handles:
+ * 1) deletion of all rows for a DROP TABLE and subsequently deletion of all rows for a DROP INDEX;
+ * 2) deletion of all column values for a ALTER TABLE DROP COLUMN
+ * 3) updating the necessary rows to have an empty KV
+ */
+ long totalMutationCount = 0;
+ for (final TableRef tableRef : tableRefs) {
+ Scan scan = new Scan();
+ scan.setAttribute(UngroupedAggregateRegionObserver.UNGROUPED_AGG, QueryConstants.TRUE);
+ SelectStatement select = SelectStatement.COUNT_ONE;
+ // We need to use this tableRef
+ ColumnResolver resolver = new ColumnResolver() {
+ @Override
+ public List<TableRef> getTables() {
+ return Collections.singletonList(tableRef);
+ }
+ @Override
+ public ColumnRef resolveColumn(String schemaName, String tableName, String colName) throws SQLException {
+ PColumn column = tableName != null
+ ? tableRef.getTable().getColumnFamily(tableName).getColumn(colName)
+ : tableRef.getTable().getColumn(colName);
+ return new ColumnRef(tableRef, column.getPosition());
+ }
+ };
+ StatementContext context = new StatementContext(new PhoenixStatement(connection), resolver, Collections.<Object>emptyList(), scan);
+ ScanUtil.setTimeRange(scan, timestamp);
+ if (emptyCF != null) {
+ scan.setAttribute(UngroupedAggregateRegionObserver.EMPTY_CF, emptyCF);
+ }
+ ServerCache cache = null;
+ try {
+ if (deleteList != null) {
+ if (deleteList.isEmpty()) {
+ scan.setAttribute(UngroupedAggregateRegionObserver.DELETE_AGG, QueryConstants.TRUE);
+ // In the case of a row deletion, add index metadata so mutable secondary indexing works
+ /* TODO
+ ImmutableBytesWritable ptr = context.getTempPtr();
+ tableRef.getTable().getIndexMaintainers(ptr);
+ if (ptr.getLength() > 0) {
+ IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
+ cache = client.addIndexMetadataCache(context.getScanRanges(), ptr);
+ byte[] uuidValue = cache.getId();
+ scan.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
+ }
+ */
+ } else {
+ // In the case of the empty key value column family changing, do not send the index
+ // metadata, as we're currently managing this from the client. It's possible for the
+ // data empty column family to stay the same, while the index empty column family
+ // changes.
+ PColumn column = deleteList.get(0);
+ if (emptyCF == null) {
+ scan.addColumn(column.getFamilyName().getBytes(), column.getName().getBytes());
+ }
+ scan.setAttribute(UngroupedAggregateRegionObserver.DELETE_CF, column.getFamilyName().getBytes());
+ scan.setAttribute(UngroupedAggregateRegionObserver.DELETE_CQ, column.getName().getBytes());
+ }
+ }
+ List<byte[]> columnFamilies = Lists.newArrayListWithExpectedSize(tableRef.getTable().getColumnFamilies().size());
+ if (projectCF == null) {
+ for (PColumnFamily family : tableRef.getTable().getColumnFamilies()) {
+ columnFamilies.add(family.getName().getBytes());
+ }
+ } else {
+ columnFamilies.add(projectCF);
+ }
+ // Need to project all column families into the scan, since we haven't yet created our empty key value
+ RowProjector projector = ProjectionCompiler.compile(context, SelectStatement.COUNT_ONE, GroupBy.EMPTY_GROUP_BY);
+ // Explicitly project these column families and don't project the empty key value,
+ // since at this point we haven't added the empty key value everywhere.
+ if (columnFamilies != null) {
+ scan.getFamilyMap().clear();
+ for (byte[] family : columnFamilies) {
+ scan.addFamily(family);
+ }
+ projector = new RowProjector(projector,false);
+ }
+ WhereCompiler.compile(context, select); // Push where clause into scan
+ QueryPlan plan = new AggregatePlan(context, select, tableRef, projector, null, OrderBy.EMPTY_ORDER_BY, null, GroupBy.EMPTY_GROUP_BY, null);
+ ResultIterator iterator = plan.iterator();
+ try {
+ Tuple row = iterator.next();
+ ImmutableBytesWritable ptr = context.getTempPtr();
+ totalMutationCount += (Long)projector.getColumnProjector(0).getValue(row, PDataType.LONG, ptr);
+ } catch (SQLException e) {
+ sqlE = e;
+ } finally {
+ try {
+ iterator.close();
+ } catch (SQLException e) {
+ if (sqlE == null) {
+ sqlE = e;
+ } else {
+ sqlE.setNextException(e);
+ }
+ } finally {
+ if (sqlE != null) {
+ throw sqlE;
+ }
+ }
+ }
+ } finally {
+ if (cache != null) { // Remove server cache if there is one
+ cache.close();
+ }
+ }
+
+ }
+ final long count = totalMutationCount;
+ return new MutationState(1, connection) {
+ @Override
+ public long getUpdateCount() {
+ return count;
+ }
+ };
+ } finally {
+ if (!wasAutoCommit) connection.setAutoCommit(wasAutoCommit);
+ }
+ }
+ };
+ }
+}