You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ha...@apache.org on 2015/02/19 19:41:05 UTC
svn commit: r1660968 - in /hive/branches/cbo:
common/src/java/org/apache/hadoop/hive/conf/
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/optimizer/
ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/ ql/src/java...
Author: hashutosh
Date: Thu Feb 19 18:41:04 2015
New Revision: 1660968
URL: http://svn.apache.org/r1660968
Log:
HIVE-9581 : CBO (Calcite Return Path): Translate Join to Hive Op [CBO branch] (Jesus Camacho Rodriguez)
Added:
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveJoinAddNotNullRule.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java
Modified:
hive/branches/cbo/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/RowSchema.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveCalciteUtil.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/reloperators/HiveTableScan.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ExprNodeConverter.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/JoinDesc.java
Modified: hive/branches/cbo/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java (original)
+++ hive/branches/cbo/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java Thu Feb 19 18:41:04 2015
@@ -695,6 +695,7 @@ public class HiveConf extends Configurat
// CBO related
HIVE_CBO_ENABLED("hive.cbo.enable", true, "Flag to control enabling Cost Based Optimizations using Calcite framework."),
+ HIVE_CBO_RETPATH_HIVEOP("hive.cbo.returnpath.hiveop", false, "Flag to control calcite plan to hive operator conversion"),
// hive.mapjoin.bucket.cache.size has been replaced by hive.smbjoin.cache.row,
// need to remove by hive .13. Also, do not change default (see SMB operator)
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/RowSchema.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/RowSchema.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/RowSchema.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/RowSchema.java Thu Feb 19 18:41:04 2015
@@ -22,6 +22,7 @@ import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
+import java.util.List;
import java.util.Set;
/**
@@ -102,6 +103,14 @@ public class RowSchema implements Serial
return tableNames;
}
+ public List<String> getColumnNames() {
+ List<String> columnNames = new ArrayList<String>();
+ for (ColumnInfo var : this.signature) {
+ columnNames.add(var.getInternalName());
+ }
+ return columnNames;
+ }
+
@Override
public boolean equals(Object obj) {
if (!(obj instanceof RowSchema) || (obj == null)) {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java Thu Feb 19 18:41:04 2015
@@ -528,14 +528,15 @@ public final class ColumnPrunerProcFacto
Operator<? extends OperatorDesc> child = op.getChildOperators().get(0);
- List<String> childCols;
+ List<String> childCols = null;
if (child instanceof CommonJoinOperator) {
- childCols = cppCtx.getJoinPrunedColLists().get(child)
- .get((byte) conf.getTag());
+ childCols = cppCtx.getJoinPrunedColLists().get(child) == null
+ ? null : cppCtx.getJoinPrunedColLists().get(child)
+ .get((byte) conf.getTag());
} else {
childCols = cppCtx.getPrunedColList(child);
-
}
+
List<ExprNodeDesc> valCols = conf.getValueCols();
List<String> valColNames = conf.getOutputValueColumnNames();
@@ -746,6 +747,7 @@ public final class ColumnPrunerProcFacto
conf.setOutputColumnNames(newOutputColumnNames);
handleChildren(op, cols, cppCtx);
}
+
return null;
}
@@ -968,16 +970,16 @@ public final class ColumnPrunerProcFacto
.getChildOperators();
LOG.info("JOIN " + op.getIdentifier() + " oldExprs: " + conf.getExprs());
+
List<String> childColLists = cppCtx.genColLists(op);
if (childColLists == null) {
- return;
- }
-
+ return;
+ }
- Map<Byte, List<String>> prunedColLists = new HashMap<Byte, List<String>>();
- for (byte tag : conf.getTagOrder()) {
- prunedColLists.put(tag, new ArrayList<String>());
- }
+ Map<Byte, List<String>> prunedColLists = new HashMap<Byte, List<String>>();
+ for (byte tag : conf.getTagOrder()) {
+ prunedColLists.put(tag, new ArrayList<String>());
+ }
//add the columns in join filters
Set<Map.Entry<Byte, List<ExprNodeDesc>>> filters =
@@ -1073,6 +1075,7 @@ public final class ColumnPrunerProcFacto
}
LOG.info("JOIN " + op.getIdentifier() + " newExprs: " + conf.getExprs());
+
op.setColumnExprMap(newColExprMap);
conf.setOutputColumnNames(outputCols);
op.getSchema().setSignature(rs);
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/NonBlockingOpDeDupProc.java Thu Feb 19 18:41:04 2015
@@ -57,11 +57,9 @@ public class NonBlockingOpDeDupProc impl
@Override
public ParseContext transform(ParseContext pctx) throws SemanticException {
- String SEL = SelectOperator.getOperatorName();
String FIL = FilterOperator.getOperatorName();
Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
- opRules.put(new RuleRegExp("R1", SEL + "%" + SEL + "%"), new SelectDedup(pctx));
- opRules.put(new RuleRegExp("R2", FIL + "%" + FIL + "%"), new FilterDedup());
+ opRules.put(new RuleRegExp("R1", FIL + "%" + FIL + "%"), new FilterDedup());
Dispatcher disp = new DefaultRuleDispatcher(null, opRules, null);
GraphWalker ogw = new DefaultGraphWalker(disp);
@@ -72,8 +70,12 @@ public class NonBlockingOpDeDupProc impl
return pctx;
}
+ @Deprecated
private class SelectDedup implements NodeProcessor {
+ // This is taken care of now by
+ // {@link org.apache.hadoop.hive.ql.optimizer.IdentityProjectRemover}
+
private ParseContext pctx;
public SelectDedup (ParseContext pctx) {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveCalciteUtil.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveCalciteUtil.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveCalciteUtil.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/HiveCalciteUtil.java Thu Feb 19 18:41:04 2015
@@ -28,6 +28,7 @@ import java.util.Set;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.RelOptUtil.InputReferencedVisitor;
import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.RelFactories.ProjectFactory;
import org.apache.calcite.rel.core.Sort;
import org.apache.calcite.rel.type.RelDataTypeField;
@@ -42,13 +43,16 @@ import org.apache.calcite.sql.validate.S
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.calcite.util.Pair;
import org.apache.calcite.util.Util;
-import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveJoin;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveProject;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
@@ -311,11 +315,11 @@ public class HiveCalciteUtil {
return this.mapOfProjIndxInJoinSchemaToLeafPInfo;
}
- public static JoinPredicateInfo constructJoinPredicateInfo(HiveJoin j) {
+ public static JoinPredicateInfo constructJoinPredicateInfo(Join j) {
return constructJoinPredicateInfo(j, j.getCondition());
}
- public static JoinPredicateInfo constructJoinPredicateInfo(HiveJoin j, RexNode predicate) {
+ public static JoinPredicateInfo constructJoinPredicateInfo(Join j, RexNode predicate) {
JoinPredicateInfo jpi = null;
JoinLeafPredicateInfo jlpi = null;
List<JoinLeafPredicateInfo> equiLPIList = new ArrayList<JoinLeafPredicateInfo>();
@@ -424,6 +428,16 @@ public class HiveCalciteUtil {
.copyOf(projsFromRightPartOfJoinKeysInJoinSchema);
}
+ public List<RexNode> getJoinKeyExprs(int input) {
+ if (input == 0) {
+ return this.joinKeyExprsFromLeft;
+ }
+ if (input == 1) {
+ return this.joinKeyExprsFromRight;
+ }
+ return null;
+ }
+
public List<RexNode> getJoinKeyExprsFromLeft() {
return this.joinKeyExprsFromLeft;
}
@@ -453,7 +467,7 @@ public class HiveCalciteUtil {
return this.projsFromRightPartOfJoinKeysInJoinSchema;
}
- private static JoinLeafPredicateInfo constructJoinLeafPredicateInfo(HiveJoin j, RexNode pe) {
+ private static JoinLeafPredicateInfo constructJoinLeafPredicateInfo(Join j, RexNode pe) {
JoinLeafPredicateInfo jlpi = null;
List<Integer> filterNulls = new ArrayList<Integer>();
List<RexNode> joinKeyExprsFromLeft = new ArrayList<RexNode>();
@@ -551,4 +565,53 @@ public class HiveCalciteUtil {
return deterministic;
}
+
+ public static ImmutableMap<Integer, ColumnInfo> getColInfoMap(List<ColumnInfo> hiveCols,
+ int startIndx) {
+ Builder<Integer, ColumnInfo> bldr = ImmutableMap.<Integer, ColumnInfo> builder();
+
+ int indx = startIndx;
+ for (ColumnInfo ci : hiveCols) {
+ bldr.put(indx, ci);
+ indx++;
+ }
+
+ return bldr.build();
+ }
+
+ public static ImmutableMap<Integer, VirtualColumn> shiftVColsMap(Map<Integer, VirtualColumn> hiveVCols,
+ int shift) {
+ Builder<Integer, VirtualColumn> bldr = ImmutableMap.<Integer, VirtualColumn> builder();
+
+ for (Integer pos : hiveVCols.keySet()) {
+ bldr.put(shift + pos, hiveVCols.get(pos));
+ }
+
+ return bldr.build();
+ }
+
+ public static ImmutableMap<Integer, VirtualColumn> getVColsMap(List<VirtualColumn> hiveVCols,
+ int startIndx) {
+ Builder<Integer, VirtualColumn> bldr = ImmutableMap.<Integer, VirtualColumn> builder();
+
+ int indx = startIndx;
+ for (VirtualColumn vc : hiveVCols) {
+ bldr.put(indx, vc);
+ indx++;
+ }
+
+ return bldr.build();
+ }
+
+ public static ImmutableMap<String, Integer> getColNameIndxMap(List<FieldSchema> tableFields) {
+ Builder<String, Integer> bldr = ImmutableMap.<String, Integer> builder();
+
+ int indx = 0;
+ for (FieldSchema fs : tableFields) {
+ bldr.put(fs.getName(), indx);
+ indx++;
+ }
+
+ return bldr.build();
+ }
}
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/RelOptHiveTable.java Thu Feb 19 18:41:04 2015
@@ -31,6 +31,7 @@ import org.apache.calcite.plan.RelOptUti
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.logical.LogicalTableScan;
import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.commons.logging.Log;
@@ -41,6 +42,7 @@ import org.apache.hadoop.hive.ql.exec.Co
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
+import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
import org.apache.hadoop.hive.ql.optimizer.calcite.translator.ExprNodeConverter;
import org.apache.hadoop.hive.ql.optimizer.ppr.PartitionPruner;
import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
@@ -51,15 +53,16 @@ import org.apache.hadoop.hive.ql.stats.S
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableMap.Builder;
public class RelOptHiveTable extends RelOptAbstractTable {
private final Table hiveTblMetadata;
private final String tblAlias;
private final ImmutableList<ColumnInfo> hiveNonPartitionCols;
+ private final ImmutableList<ColumnInfo> hivePartitionCols;
private final ImmutableMap<Integer, ColumnInfo> hiveNonPartitionColsMap;
private final ImmutableMap<Integer, ColumnInfo> hivePartitionColsMap;
- private final int noOfProjs;
+ private final ImmutableList<VirtualColumn> hiveVirtualCols;
+ private final int noOfNonVirtualCols;
final HiveConf hiveConf;
private double rowCount = -1;
@@ -67,37 +70,66 @@ public class RelOptHiveTable extends Rel
PrunedPartitionList partitionList;
Map<String, PrunedPartitionList> partitionCache;
AtomicInteger noColsMissingStats;
+ private final String qbID;
- protected static final Log LOG = LogFactory
- .getLog(RelOptHiveTable.class
- .getName());
-
- public RelOptHiveTable(RelOptSchema calciteSchema, String qualifiedTblName, String tblAlias, RelDataType rowType,
- Table hiveTblMetadata, List<ColumnInfo> hiveNonPartitionCols,
- List<ColumnInfo> hivePartitionCols, HiveConf hconf, Map<String, PrunedPartitionList> partitionCache, AtomicInteger noColsMissingStats) {
+ protected static final Log LOG = LogFactory
+ .getLog(RelOptHiveTable.class
+ .getName());
+
+ public RelOptHiveTable(RelOptSchema calciteSchema, String qualifiedTblName, String tblAlias,
+ RelDataType rowType, Table hiveTblMetadata, List<ColumnInfo> hiveNonPartitionCols,
+ List<ColumnInfo> hivePartitionCols, List<VirtualColumn> hiveVirtualCols, HiveConf hconf,
+ Map<String, PrunedPartitionList> partitionCache, AtomicInteger noColsMissingStats,
+ String qbID) {
super(calciteSchema, qualifiedTblName, rowType);
this.hiveTblMetadata = hiveTblMetadata;
this.tblAlias = tblAlias;
this.hiveNonPartitionCols = ImmutableList.copyOf(hiveNonPartitionCols);
- this.hiveNonPartitionColsMap = getColInfoMap(hiveNonPartitionCols, 0);
- this.hivePartitionColsMap = getColInfoMap(hivePartitionCols, hiveNonPartitionColsMap.size());
- this.noOfProjs = hiveNonPartitionCols.size() + hivePartitionCols.size();
+ this.hiveNonPartitionColsMap = HiveCalciteUtil.getColInfoMap(hiveNonPartitionCols, 0);
+ this.hivePartitionCols = ImmutableList.copyOf(hivePartitionCols);
+ this.hivePartitionColsMap = HiveCalciteUtil.getColInfoMap(hivePartitionCols, hiveNonPartitionColsMap.size());
+ this.noOfNonVirtualCols = hiveNonPartitionCols.size() + hivePartitionCols.size();
+ this.hiveVirtualCols = ImmutableList.copyOf(hiveVirtualCols);
this.hiveConf = hconf;
this.partitionCache = partitionCache;
this.noColsMissingStats = noColsMissingStats;
+ this.qbID = qbID;
}
- private static ImmutableMap<Integer, ColumnInfo> getColInfoMap(List<ColumnInfo> hiveCols,
- int startIndx) {
- Builder<Integer, ColumnInfo> bldr = ImmutableMap.<Integer, ColumnInfo> builder();
-
- int indx = startIndx;
- for (ColumnInfo ci : hiveCols) {
- bldr.put(indx, ci);
- indx++;
+ public RelOptHiveTable copy(RelDataType newRowType) {
+ // 1. Build map of column name to col index of original schema
+ // Assumption: Hive Table can not contain duplicate column names
+ Map<String, Integer> nameToColIndxMap = new HashMap<String, Integer>();
+ for (RelDataTypeField f : this.rowType.getFieldList()) {
+ nameToColIndxMap.put(f.getName(), f.getIndex());
+ }
+
+ // 2. Build nonPart/Part/Virtual column info for new RowSchema
+ List<ColumnInfo> newHiveNonPartitionCols = new ArrayList<ColumnInfo>();
+ List<ColumnInfo> newHivePartitionCols = new ArrayList<ColumnInfo>();
+ List<VirtualColumn> newHiveVirtualCols = new ArrayList<VirtualColumn>();
+ Map<Integer, VirtualColumn> virtualColInfoMap = HiveCalciteUtil.getVColsMap(this.hiveVirtualCols,
+ this.noOfNonVirtualCols);
+ Integer originalColIndx;
+ ColumnInfo cInfo;
+ VirtualColumn vc;
+ for (RelDataTypeField f : newRowType.getFieldList()) {
+ originalColIndx = nameToColIndxMap.get(f.getName());
+ if ((cInfo = hiveNonPartitionColsMap.get(originalColIndx)) != null) {
+ newHiveNonPartitionCols.add(new ColumnInfo(cInfo));
+ } else if ((cInfo = hivePartitionColsMap.get(originalColIndx)) != null) {
+ newHivePartitionCols.add(new ColumnInfo(cInfo));
+ } else if ((vc = virtualColInfoMap.get(originalColIndx)) != null) {
+ newHiveVirtualCols.add(vc);
+ } else {
+ throw new RuntimeException("Copy encountered a column not seen in original TS");
+ }
}
- return bldr.build();
+ // 3. Build new Table
+ return new RelOptHiveTable(this.schema, this.name, this.tblAlias, newRowType,
+ this.hiveTblMetadata, newHiveNonPartitionCols, newHivePartitionCols, newHiveVirtualCols,
+ this.hiveConf, this.partitionCache, this.noColsMissingStats, qbID);
}
@Override
@@ -119,13 +151,13 @@ public class RelOptHiveTable extends Rel
public double getRowCount() {
if (rowCount == -1) {
if (null == partitionList) {
- // we are here either unpartitioned table or partitioned table with no predicates
+ // we are here either unpartitioned table or partitioned table with no
+ // predicates
computePartitionList(hiveConf, null);
}
if (hiveTblMetadata.isPartitioned()) {
- List<Long> rowCounts = StatsUtils.getBasicStatForPartitions(
- hiveTblMetadata, partitionList.getNotDeniedPartns(),
- StatsSetupConst.ROW_COUNT);
+ List<Long> rowCounts = StatsUtils.getBasicStatForPartitions(hiveTblMetadata,
+ partitionList.getNotDeniedPartns(), StatsSetupConst.ROW_COUNT);
rowCount = StatsUtils.getSumIgnoreNegatives(rowCounts);
} else {
@@ -144,8 +176,10 @@ public class RelOptHiveTable extends Rel
}
public String getTableAlias() {
- // NOTE: Calcite considers tbls to be equal if their names are the same. Hence
- // we need to provide Calcite the fully qualified table name (dbname.tblname)
+ // NOTE: Calcite considers tbls to be equal if their names are the same.
+ // Hence
+ // we need to provide Calcite the fully qualified table name
+ // (dbname.tblname)
// and not the user provided aliases.
// However in HIVE DB name can not appear in select list; in case of join
// where table names differ only in DB name, Hive would require user
@@ -173,16 +207,21 @@ public class RelOptHiveTable extends Rel
public void computePartitionList(HiveConf conf, RexNode pruneNode) {
try {
- if (!hiveTblMetadata.isPartitioned() || pruneNode == null || InputFinder.bits(pruneNode).length() == 0 ) {
- // there is no predicate on partitioning column, we need all partitions in this case.
- partitionList = PartitionPruner.prune(hiveTblMetadata, null, conf, getName(), partitionCache);
+ if (!hiveTblMetadata.isPartitioned() || pruneNode == null
+ || InputFinder.bits(pruneNode).length() == 0) {
+ // there is no predicate on partitioning column, we need all partitions
+ // in this case.
+ partitionList = PartitionPruner.prune(hiveTblMetadata, null, conf, getName(),
+ partitionCache);
return;
}
// We have valid pruning expressions, only retrieve qualifying partitions
- ExprNodeDesc pruneExpr = pruneNode.accept(new ExprNodeConverter(getName(), getRowType(), true, getRelOptSchema().getTypeFactory()));
+ ExprNodeDesc pruneExpr = pruneNode
+ .accept(new ExprNodeConverter(getName(), getRowType(), true));
- partitionList = PartitionPruner.prune(hiveTblMetadata, pruneExpr, conf, getName(), partitionCache);
+ partitionList = PartitionPruner.prune(hiveTblMetadata, pruneExpr, conf, getName(),
+ partitionCache);
} catch (HiveException he) {
throw new RuntimeException(he);
}
@@ -289,10 +328,10 @@ public class RelOptHiveTable extends Rel
if (colNamesFailedStats.isEmpty() && !partColNamesThatRqrStats.isEmpty()) {
ColStatistics cStats = null;
for (int i = 0; i < partColNamesThatRqrStats.size(); i++) {
- cStats = new ColStatistics(hiveTblMetadata.getTableName(),
- partColNamesThatRqrStats.get(i), hivePartitionColsMap.get(
- partColIndxsThatRqrStats.get(i)).getTypeName());
- cStats.setCountDistint(getDistinctCount(partitionList.getPartitions(),partColNamesThatRqrStats.get(i)));
+ cStats = new ColStatistics(hiveTblMetadata.getTableName(), partColNamesThatRqrStats.get(i),
+ hivePartitionColsMap.get(partColIndxsThatRqrStats.get(i)).getTypeName());
+ cStats.setCountDistint(getDistinctCount(partitionList.getPartitions(),
+ partColNamesThatRqrStats.get(i)));
hiveColStatsMap.put(partColIndxsThatRqrStats.get(i), cStats);
}
}
@@ -325,7 +364,7 @@ public class RelOptHiveTable extends Rel
}
} else {
List<Integer> pILst = new ArrayList<Integer>();
- for (Integer i = 0; i < noOfProjs; i++) {
+ for (Integer i = 0; i < noOfNonVirtualCols; i++) {
pILst.add(i);
}
updateColStats(new HashSet<Integer>(pILst));
@@ -338,10 +377,8 @@ public class RelOptHiveTable extends Rel
}
/*
- * use to check if a set of columns are all partition columns.
- * true only if:
- * - all columns in BitSet are partition
- * columns.
+ * use to check if a set of columns are all partition columns. true only if: -
+ * all columns in BitSet are partition columns.
*/
public boolean containsPartitionColumnsOnly(ImmutableBitSet cols) {
@@ -352,4 +389,20 @@ public class RelOptHiveTable extends Rel
}
return true;
}
+
+ public List<VirtualColumn> getVirtualCols() {
+ return this.hiveVirtualCols;
+ }
+
+ public List<ColumnInfo> getPartColumns() {
+ return this.hivePartitionCols;
+ }
+
+ public List<ColumnInfo> getNonPartColumns() {
+ return this.hiveNonPartitionCols;
+ }
+
+ public String getQBID() {
+ return qbID;
+ }
}
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/reloperators/HiveTableScan.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/reloperators/HiveTableScan.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/reloperators/HiveTableScan.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/reloperators/HiveTableScan.java Thu Feb 19 18:41:04 2015
@@ -54,8 +54,7 @@ public class HiveTableScan extends Table
* @param table
* HiveDB table
*/
- public HiveTableScan(RelOptCluster cluster, RelTraitSet traitSet, RelOptHiveTable table,
- RelDataType rowtype) {
+ public HiveTableScan(RelOptCluster cluster, RelTraitSet traitSet, RelOptHiveTable table) {
super(cluster, TraitsUtil.getDefaultTraitSet(cluster), table);
assert getConvention() == HiveRelNode.CONVENTION;
}
@@ -66,6 +65,17 @@ public class HiveTableScan extends Table
return this;
}
+ /**
+ * Copy TableScan operator with a new Row Schema. The new Row Schema can only
+ * be a subset of this TS schema.
+ *
+ * @param rowtype
+ * @return
+ */
+ public HiveTableScan copy(RelDataType newRowtype) {
+ return new HiveTableScan(getCluster(), getTraitSet(), ((RelOptHiveTable) table).copy(newRowtype));
+ }
+
@Override
public RelOptCost computeSelfCost(RelOptPlanner planner) {
return HiveCost.FACTORY.makeZeroCost();
@@ -85,8 +95,9 @@ public class HiveTableScan extends Table
public double getRows() {
return ((RelOptHiveTable) table).getRowCount();
}
-
+
public List<ColStatistics> getColStat(List<Integer> projIndxLst) {
return ((RelOptHiveTable) table).getColStat(projIndxLst);
}
+
}
\ No newline at end of file
Added: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveJoinAddNotNullRule.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveJoinAddNotNullRule.java?rev=1660968&view=auto
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveJoinAddNotNullRule.java (added)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/rules/HiveJoinAddNotNullRule.java Thu Feb 19 18:41:04 2015
@@ -0,0 +1,193 @@
+/**
+ * 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.hadoop.hive.ql.optimizer.calcite.rules;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Join;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.RelFactories.FilterFactory;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.hadoop.hive.ql.exec.FunctionRegistry;
+import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil;
+import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil.JoinLeafPredicateInfo;
+import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil.JoinPredicateInfo;
+import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveFilter;
+import org.apache.hadoop.hive.ql.optimizer.calcite.translator.SqlFunctionConverter;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+
+import com.google.common.collect.ImmutableList;
+
+public final class HiveJoinAddNotNullRule extends RelOptRule {
+
+ private static final String NOT_NULL_FUNC_NAME = "isnotnull";
+
+ /** The singleton. */
+ public static final HiveJoinAddNotNullRule INSTANCE =
+ new HiveJoinAddNotNullRule(HiveFilter.DEFAULT_FILTER_FACTORY);
+
+ private final FilterFactory filterFactory;
+
+ //~ Constructors -----------------------------------------------------------
+
+ /**
+ * Creates an HiveJoinAddNotNullRule.
+ */
+ public HiveJoinAddNotNullRule(FilterFactory filterFactory) {
+ super(operand(Join.class,
+ operand(RelNode.class, any()),
+ operand(RelNode.class, any())));
+ this.filterFactory = filterFactory;
+ }
+
+ //~ Methods ----------------------------------------------------------------
+
+ public void onMatch(RelOptRuleCall call) {
+ final Join join = call.rel(0);
+ RelNode leftInput = call.rel(1);
+ RelNode rightInput = call.rel(2);
+
+ if (join.getJoinType() != JoinRelType.INNER) {
+ return;
+ }
+
+ if (join.getCondition().isAlwaysTrue()) {
+ return;
+ }
+
+ JoinPredicateInfo joinPredInfo =
+ HiveCalciteUtil.JoinPredicateInfo.constructJoinPredicateInfo(join);
+
+ Set<Integer> joinLeftKeyPositions = new HashSet<Integer>();
+ Set<Integer> joinRightKeyPositions = new HashSet<Integer>();
+ for (int i = 0; i < joinPredInfo.getEquiJoinPredicateElements().size(); i++) {
+ JoinLeafPredicateInfo joinLeafPredInfo = joinPredInfo.
+ getEquiJoinPredicateElements().get(i);
+ joinLeftKeyPositions.addAll(joinLeafPredInfo.getProjsFromLeftPartOfJoinKeysInChildSchema());
+ joinRightKeyPositions.addAll(joinLeafPredInfo.getProjsFromRightPartOfJoinKeysInChildSchema());
+ }
+
+ // Build not null conditions
+ final RelOptCluster cluster = join.getCluster();
+ final RexBuilder rexBuilder = join.getCluster().getRexBuilder();
+
+ final Map<String,RexNode> newLeftConditions = getNotNullConditions(cluster,
+ rexBuilder, leftInput, joinLeftKeyPositions);
+ final Map<String,RexNode> newRightConditions = getNotNullConditions(cluster,
+ rexBuilder, rightInput, joinRightKeyPositions);
+
+ // Nothing will be added to the expression
+ if (newLeftConditions == null && newRightConditions == null) {
+ return;
+ }
+
+ if (newLeftConditions != null) {
+ if (leftInput instanceof HiveFilter) {
+ leftInput = leftInput.getInput(0);
+ }
+ leftInput = createHiveFilterConjunctiveCondition(filterFactory, rexBuilder,
+ leftInput, newLeftConditions.values());
+ }
+ if (newRightConditions != null) {
+ if (rightInput instanceof HiveFilter) {
+ rightInput = rightInput.getInput(0);
+ }
+ rightInput = createHiveFilterConjunctiveCondition(filterFactory, rexBuilder,
+ rightInput, newRightConditions.values());
+ }
+
+ Join newJoin = join.copy(join.getTraitSet(), join.getCondition(),
+ leftInput, rightInput, join.getJoinType(), join.isSemiJoinDone());
+
+ call.getPlanner().onCopy(join, newJoin);
+
+ call.transformTo(newJoin);
+ }
+
+ private static Map<String,RexNode> getNotNullConditions(RelOptCluster cluster,
+ RexBuilder rexBuilder, RelNode input, Set<Integer> inputKeyPositions) {
+
+ boolean added = false;
+
+ final RelDataType returnType = cluster.getTypeFactory().
+ createSqlType(SqlTypeName.BOOLEAN);
+
+ final Map<String,RexNode> newConditions;
+ if (input instanceof HiveFilter) {
+ newConditions = splitCondition(((HiveFilter) input).getCondition());
+ }
+ else {
+ newConditions = new HashMap<String,RexNode>();
+ }
+ for (int pos : inputKeyPositions) {
+ try {
+ RelDataType keyType = input.getRowType().getFieldList().get(pos).getType();
+ SqlOperator funcCall = SqlFunctionConverter.getCalciteOperator(NOT_NULL_FUNC_NAME,
+ FunctionRegistry.getFunctionInfo(NOT_NULL_FUNC_NAME).getGenericUDF(),
+ ImmutableList.of(keyType), returnType);
+ RexNode cond = rexBuilder.makeCall(funcCall, rexBuilder.makeInputRef(input, pos));
+ String digest = cond.toString();
+ if (!newConditions.containsKey(digest)) {
+ newConditions.put(digest,cond);
+ added = true;
+ }
+ } catch (SemanticException e) {
+ throw new AssertionError(e.getMessage());
+ }
+ }
+ // Nothing will be added to the expression
+ if (!added) {
+ return null;
+ }
+ return newConditions;
+ }
+
+ private static Map<String,RexNode> splitCondition(RexNode condition) {
+ Map<String,RexNode> newConditions = new HashMap<String,RexNode>();
+ if (condition.getKind() == SqlKind.AND) {
+ for (RexNode node : ((RexCall) condition).getOperands()) {
+ newConditions.put(node.toString(), node);
+ }
+ }
+ else {
+ newConditions.put(condition.toString(), condition);
+ }
+ return newConditions;
+ }
+
+ private static RelNode createHiveFilterConjunctiveCondition(FilterFactory filterFactory,
+ RexBuilder rexBuilder, RelNode input, Collection<RexNode> conditions) {
+ final RexNode newCondition = RexUtil.composeConjunction(rexBuilder, conditions, false);
+ return filterFactory.createFilter(input, newCondition);
+ }
+}
\ No newline at end of file
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ExprNodeConverter.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ExprNodeConverter.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ExprNodeConverter.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/ExprNodeConverter.java Thu Feb 19 18:41:04 2015
@@ -33,15 +33,12 @@ import org.apache.hadoop.hive.ql.plan.Ex
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.type.RelDataTypeFactory;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexVisitorImpl;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.type.SqlTypeUtil;
/*
* convert a RexNode to an ExprNodeDesc
@@ -51,22 +48,12 @@ public class ExprNodeConverter extends R
RelDataType rType;
String tabAlias;
boolean partitioningExpr;
- private final RelDataTypeFactory dTFactory;
- public ExprNodeConverter(String tabAlias, RelDataType rType, boolean partitioningExpr, RelDataTypeFactory dTFactory) {
+ public ExprNodeConverter(String tabAlias, RelDataType rType, boolean partitioningExpr) {
super(true);
- /*
- * hb: 6/25/14 for now we only support expressions that only contain
- * partition cols. there is no use case for supporting generic expressions.
- * for supporting generic exprs., we need to give the converter information
- * on whether a column is a partition column or not, whether a column is a
- * virtual column or not.
- */
- assert partitioningExpr == true;
this.tabAlias = tabAlias;
this.rType = rType;
this.partitioningExpr = partitioningExpr;
- this.dTFactory = dTFactory;
}
@Override
@@ -76,6 +63,9 @@ public class ExprNodeConverter extends R
partitioningExpr);
}
+ /**
+ * TODO: Handle 1) cast 2) Field Access 3) Windowing Over() 4, Windowing Agg Call
+ */
@Override
public ExprNodeDesc visitCall(RexCall call) {
ExprNodeGenericFuncDesc gfDesc = null;
@@ -90,15 +80,9 @@ public class ExprNodeConverter extends R
args.add(operand.accept(this));
}
- // If Call is a redundant cast then bail out. Ex: cast(true)BOOLEAN
- if (call.isA(SqlKind.CAST)
- && (call.operands.size() == 1)
- && SqlTypeUtil.equalSansNullability(dTFactory, call.getType(),
- call.operands.get(0).getType())) {
- return args.get(0);
- } else if (ASTConverter.isFlat(call)) {
- // If Expr is flat (and[p,q,r,s] or[p,q,r,s]) then recursively build the
- // exprnode
+ // If Expr is flat (and[p,q,r,s] or[p,q,r,s]) then recursively build the
+ // exprnode
+ if (ASTConverter.isFlat(call)) {
ArrayList<ExprNodeDesc> tmpExprArgs = new ArrayList<ExprNodeDesc>();
tmpExprArgs.addAll(args.subList(0, 2));
gfDesc = new ExprNodeGenericFuncDesc(TypeConverter.convert(call.getType()),
@@ -123,6 +107,9 @@ public class ExprNodeConverter extends R
return gfDesc;
}
+ /**
+ * TODO: 1. Handle NULL
+ */
@Override
public ExprNodeDesc visitLiteral(RexLiteral literal) {
RelDataType lType = literal.getType();
Added: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java?rev=1660968&view=auto
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java (added)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/calcite/translator/HiveOpConverter.java Thu Feb 19 18:41:04 2015
@@ -0,0 +1,498 @@
+/**
+ * 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.hadoop.hive.ql.optimizer.calcite.translator;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.SemiJoin;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.Pair;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.exec.JoinOperator;
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.OperatorFactory;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
+import org.apache.hadoop.hive.ql.exec.RowSchema;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.io.AcidUtils.Operation;
+import org.apache.hadoop.hive.ql.metadata.VirtualColumn;
+import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil;
+import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil.JoinLeafPredicateInfo;
+import org.apache.hadoop.hive.ql.optimizer.calcite.HiveCalciteUtil.JoinPredicateInfo;
+import org.apache.hadoop.hive.ql.optimizer.calcite.reloperators.HiveJoin;
+import org.apache.hadoop.hive.ql.parse.JoinCond;
+import org.apache.hadoop.hive.ql.parse.JoinType;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeDescUtils;
+import org.apache.hadoop.hive.ql.plan.JoinCondDesc;
+import org.apache.hadoop.hive.ql.plan.JoinDesc;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+import org.apache.hadoop.hive.ql.plan.PlanUtils;
+import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+public class HiveOpConverter {
+
+ private static final Log LOG = LogFactory.getLog(HiveOpConverter.class);
+
+ public static enum HIVEAGGOPMODE {
+ NO_SKEW_NO_MAP_SIDE_AGG, // Corresponds to SemAnalyzer genGroupByPlan1MR
+ SKEW_NO_MAP_SIDE_AGG, // Corresponds to SemAnalyzer genGroupByPlan2MR
+ NO_SKEW_MAP_SIDE_AGG, // Corresponds to SemAnalyzer
+ // genGroupByPlanMapAggrNoSkew
+ SKEW_MAP_SIDE_AGG // Corresponds to SemAnalyzer genGroupByPlanMapAggr2MR
+ };
+
+ // TODO: remove this after stashing only rqd pieces from opconverter
+ private final Map<String, Operator<? extends OperatorDesc>> topOps;
+ private final HIVEAGGOPMODE aggMode;
+ private final boolean strictMode;
+
+ public static HIVEAGGOPMODE getAggOPMode(HiveConf hc) {
+ HIVEAGGOPMODE aggOpMode = HIVEAGGOPMODE.NO_SKEW_NO_MAP_SIDE_AGG;
+
+ if (hc.getBoolVar(HiveConf.ConfVars.HIVEMAPSIDEAGGREGATE)) {
+ if (!hc.getBoolVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)) {
+ aggOpMode = HIVEAGGOPMODE.NO_SKEW_MAP_SIDE_AGG;
+ } else {
+ aggOpMode = HIVEAGGOPMODE.SKEW_MAP_SIDE_AGG;
+ }
+ } else if (hc.getBoolVar(HiveConf.ConfVars.HIVEGROUPBYSKEW)) {
+ aggOpMode = HIVEAGGOPMODE.SKEW_NO_MAP_SIDE_AGG;
+ }
+
+ return aggOpMode;
+ }
+
+ public HiveOpConverter(Map<String, Operator<? extends OperatorDesc>> topOps,
+ HIVEAGGOPMODE aggMode, boolean strictMode) {
+ this.topOps = topOps;
+ this.aggMode = aggMode;
+ this.strictMode = strictMode;
+ }
+
+ private class OpAttr {
+ private final String tabAlias;
+ ImmutableList<Operator> inputs;
+ ImmutableMap<Integer, VirtualColumn> vcolMap;
+
+ private OpAttr(String tabAlias, Map<Integer, VirtualColumn> vcolMap, Operator... inputs) {
+ this.tabAlias = tabAlias;
+ this.vcolMap = ImmutableMap.copyOf(vcolMap);
+ this.inputs = ImmutableList.copyOf(inputs);
+ }
+
+ private OpAttr clone(Operator... inputs) {
+ return new OpAttr(tabAlias, this.vcolMap, inputs);
+ }
+ }
+
+ public Operator convert(RelNode root) throws SemanticException {
+ OpAttr opAf = dispatch(root);
+ return opAf.inputs.get(0);
+ }
+
+ OpAttr dispatch(RelNode rn) throws SemanticException {
+ if (rn instanceof HiveJoin) {
+ return visit((HiveJoin) rn);
+ } else if (rn instanceof SemiJoin) {
+ SemiJoin sj = (SemiJoin) rn;
+ HiveJoin hj = HiveJoin.getJoin(sj.getCluster(), sj.getLeft(), sj.getRight(),
+ sj.getCondition(), sj.getJoinType(), true);
+ return visit(hj);
+ }
+ LOG.error(rn.getClass().getCanonicalName() + "operator translation not supported"
+ + " yet in return path.");
+ return null;
+ }
+
+ OpAttr visit(HiveJoin joinRel) throws SemanticException {
+ // 1. Convert inputs
+ OpAttr[] inputs = new OpAttr[joinRel.getInputs().size()];
+ for (int i=0; i<inputs.length; i++) {
+ inputs[i] = dispatch(joinRel.getInput(i));
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Translating operator rel#" + joinRel.getId() + ":" + joinRel.getRelTypeName() +
+ " with row type: [" + joinRel.getRowType() + "]");
+ }
+
+ // 2. Convert join condition
+ JoinPredicateInfo joinPredInfo = JoinPredicateInfo.constructJoinPredicateInfo(joinRel);
+
+ // 3. Extract join keys from condition
+ ExprNodeDesc[][] joinKeys = extractJoinKeys(joinPredInfo, joinRel.getInputs());
+
+ // 4. For each input, we generate the corresponding ReduceSink child
+ List<Operator<?>> children = new ArrayList<Operator<?>>();
+ for (int i = 0; i < inputs.length; i++) {
+ // Generate a ReduceSink operator for each join child
+ ReduceSinkOperator child = genReduceSink(inputs[i].inputs.get(0), joinKeys[i],
+ i, -1, Operation.NOT_ACID, strictMode);
+ children.add(child);
+ }
+
+ // 5. Generate Join operator
+ JoinOperator joinOp = genJoin(joinRel, joinPredInfo, children, joinKeys);
+
+ // 6. TODO: Extract condition for non-equi join elements (if any) and add it
+
+ // 7. Virtual columns
+ Map<Integer,VirtualColumn> vcolMap = new HashMap<Integer,VirtualColumn>();
+ vcolMap.putAll(inputs[0].vcolMap);
+ if (extractJoinType(joinRel) != JoinType.LEFTSEMI) {
+ int shift = inputs[0].inputs.get(0).getSchema().getSignature().size();
+ for (int i=1; i<inputs.length; i++) {
+ vcolMap.putAll(HiveCalciteUtil.shiftVColsMap(inputs[i].vcolMap,shift));
+ shift += inputs[i].inputs.get(0).getSchema().getSignature().size();
+ }
+ }
+
+ // 8. Return result
+ return new OpAttr(null, vcolMap, joinOp);
+ }
+
+ private static ExprNodeDesc[][] extractJoinKeys(JoinPredicateInfo joinPredInfo,
+ List<RelNode> inputs) {
+ ExprNodeDesc[][] joinKeys = new ExprNodeDesc[inputs.size()][];
+ for (int i = 0; i < inputs.size(); i++) {
+ joinKeys[i] = new ExprNodeDesc[joinPredInfo.getEquiJoinPredicateElements().size()];
+ for (int j = 0; j < joinPredInfo.getEquiJoinPredicateElements().size(); j++) {
+ JoinLeafPredicateInfo joinLeafPredInfo = joinPredInfo.getEquiJoinPredicateElements().get(j);
+ RexNode key = joinLeafPredInfo.getJoinKeyExprs(j).get(0);
+ joinKeys[i][j] = convertToExprNode(key, inputs.get(j), null);
+ }
+ }
+ return joinKeys;
+ }
+
+ private static ReduceSinkOperator genReduceSink(Operator<?> input, ExprNodeDesc[] keys,
+ int tag, int numReducers, Operation acidOperation,
+ boolean strictMode) throws SemanticException {
+ return genReduceSink(input, keys, tag, new ArrayList<ExprNodeDesc>(),
+ "", numReducers, acidOperation, strictMode);
+ }
+
+ @SuppressWarnings({ "rawtypes", "unchecked" })
+ private static ReduceSinkOperator genReduceSink(Operator<?> input,
+ ExprNodeDesc[] keys, int tag, ArrayList<ExprNodeDesc> partitionCols,
+ String order, int numReducers, Operation acidOperation,
+ boolean strictMode) throws SemanticException {
+ Operator dummy = Operator.createDummy(); // dummy for backtracking
+ dummy.setParentOperators(Arrays.asList(input));
+
+ ArrayList<ExprNodeDesc> reduceKeys = new ArrayList<ExprNodeDesc>();
+ ArrayList<ExprNodeDesc> reduceKeysBack = new ArrayList<ExprNodeDesc>();
+
+ // Compute join keys and store in reduceKeys
+ for (ExprNodeDesc key : keys) {
+ reduceKeys.add(key);
+ reduceKeysBack.add(ExprNodeDescUtils.backtrack(key, dummy, input));
+ }
+
+ // Walk over the input schema and copy in the output
+ ArrayList<ExprNodeDesc> reduceValues = new ArrayList<ExprNodeDesc>();
+ ArrayList<ExprNodeDesc> reduceValuesBack = new ArrayList<ExprNodeDesc>();
+ Map<String, ExprNodeDesc> colExprMap = new HashMap<String, ExprNodeDesc>();
+
+ List<ColumnInfo> inputColumns = input.getSchema().getSignature();
+ ArrayList<ColumnInfo> outputColumns = new ArrayList<ColumnInfo>();
+ List<String> outputColumnNames = new ArrayList<String>();
+ int[] index = new int[inputColumns.size()];
+ for (int i = 0; i < inputColumns.size(); i++) {
+ ColumnInfo colInfo = inputColumns.get(i);
+ String outputColName = colInfo.getInternalName();
+ ExprNodeDesc expr = new ExprNodeColumnDesc(colInfo);
+
+ // backtrack can be null when input is script operator
+ ExprNodeDesc exprBack = ExprNodeDescUtils.backtrack(expr, dummy, input);
+ int kindex = exprBack == null ? -1 : ExprNodeDescUtils.indexOf(exprBack, reduceKeysBack);
+ if (kindex >= 0) {
+ ColumnInfo newColInfo = new ColumnInfo(colInfo);
+ newColInfo.setInternalName(Utilities.ReduceField.KEY + ".reducesinkkey" + kindex);
+ newColInfo.setAlias(outputColName);
+ newColInfo.setTabAlias(colInfo.getTabAlias());
+ outputColumns.add(newColInfo);
+ index[i] = kindex;
+ continue;
+ }
+ int vindex = exprBack == null ? -1 : ExprNodeDescUtils.indexOf(exprBack, reduceValuesBack);
+ if (kindex >= 0) {
+ index[i] = -vindex - 1;
+ continue;
+ }
+ index[i] = -reduceValues.size() - 1;
+
+ reduceValues.add(expr);
+ reduceValuesBack.add(exprBack);
+
+ ColumnInfo newColInfo = new ColumnInfo(colInfo);
+ newColInfo.setInternalName(Utilities.ReduceField.VALUE + "." + outputColName);
+ newColInfo.setAlias(outputColName);
+ newColInfo.setTabAlias(colInfo.getTabAlias());
+
+ outputColumns.add(newColInfo);
+ outputColumnNames.add(outputColName);
+ }
+ dummy.setParentOperators(null);
+
+ // Use only 1 reducer if no reduce keys
+ if (reduceKeys.size() == 0) {
+ numReducers = 1;
+
+ // Cartesian product is not supported in strict mode
+ if (strictMode) {
+ throw new SemanticException(ErrorMsg.NO_CARTESIAN_PRODUCT.getMsg());
+ }
+ }
+
+ ReduceSinkDesc rsDesc;
+ if (order.isEmpty()) {
+ rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys, reduceValues,
+ outputColumnNames, false, tag, reduceKeys.size(),
+ numReducers, acidOperation);
+ } else {
+ rsDesc = PlanUtils.getReduceSinkDesc(reduceKeys, reduceValues,
+ outputColumnNames, false, tag, partitionCols,
+ order, numReducers, acidOperation);
+ }
+
+ ReduceSinkOperator rsOp = (ReduceSinkOperator) OperatorFactory.getAndMakeChild(rsDesc,
+ new RowSchema(outputColumns), input);
+
+ List<String> keyColNames = rsDesc.getOutputKeyColumnNames();
+ for (int i = 0 ; i < keyColNames.size(); i++) {
+ colExprMap.put(Utilities.ReduceField.KEY + "." + keyColNames.get(i), reduceKeys.get(i));
+ }
+ List<String> valColNames = rsDesc.getOutputValueColumnNames();
+ for (int i = 0 ; i < valColNames.size(); i++) {
+ colExprMap.put(Utilities.ReduceField.VALUE + "." + valColNames.get(i), reduceValues.get(i));
+ }
+
+ rsOp.setValueIndex(index);
+ rsOp.setColumnExprMap(colExprMap);
+ rsOp.setInputAliases(input.getSchema().getColumnNames().toArray(
+ new String[input.getSchema().getColumnNames().size()]));
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Generated " + rsOp + " with row schema: [" + rsOp.getSchema() + "]");
+ }
+
+ return rsOp;
+ }
+
+ private static JoinOperator genJoin(HiveJoin hiveJoin, JoinPredicateInfo joinPredInfo,
+ List<Operator<?>> children, ExprNodeDesc[][] joinKeys) throws SemanticException {
+
+ // Extract join type
+ JoinType joinType = extractJoinType(hiveJoin);
+
+ // NOTE: Currently binary joins only
+ JoinCondDesc[] joinCondns = new JoinCondDesc[1];
+ joinCondns[0] = new JoinCondDesc(new JoinCond(0, 1, joinType));
+
+ ArrayList<ColumnInfo> outputColumns = new ArrayList<ColumnInfo>();
+ ArrayList<String> outputColumnNames =
+ new ArrayList<String>(hiveJoin.getRowType().getFieldNames());
+ Operator<?>[] childOps = new Operator[children.size()];
+
+ Map<String, Byte> reversedExprs = new HashMap<String, Byte>();
+ HashMap<Byte, List<ExprNodeDesc>> exprMap = new HashMap<Byte, List<ExprNodeDesc>>();
+ Map<String, ExprNodeDesc> colExprMap = new HashMap<String, ExprNodeDesc>();
+ HashMap<Integer, Set<String>> posToAliasMap = new HashMap<Integer, Set<String>>();
+
+ int outputPos = 0;
+ for (int pos = 0; pos < children.size(); pos++) {
+ ReduceSinkOperator inputRS = (ReduceSinkOperator) children.get(pos);
+ if (inputRS.getNumParent() != 1) {
+ throw new SemanticException("RS should have single parent");
+ }
+ Operator<?> parent = inputRS.getParentOperators().get(0);
+ ReduceSinkDesc rsDesc = inputRS.getConf();
+
+ int[] index = inputRS.getValueIndex();
+
+ Byte tag = (byte) rsDesc.getTag();
+
+ // Semijoin
+ if (joinType == JoinType.LEFTSEMI && pos != 0) {
+ exprMap.put(tag, new ArrayList<ExprNodeDesc>());
+ childOps[pos] = inputRS;
+ continue;
+ }
+
+ List<String> keyColNames = rsDesc.getOutputKeyColumnNames();
+ List<String> valColNames = rsDesc.getOutputValueColumnNames();
+
+ posToAliasMap.put(pos, new HashSet<String>(inputRS.getSchema().getTableNames()));
+
+ Map<String, ExprNodeDesc> descriptors = buildBacktrackFromReduceSink(outputPos,
+ outputColumnNames, keyColNames, valColNames, index, parent);
+
+ List<ColumnInfo> parentColumns = parent.getSchema().getSignature();
+ for (int i = 0; i < index.length; i++) {
+ ColumnInfo info = new ColumnInfo(parentColumns.get(i));
+ info.setInternalName(outputColumnNames.get(outputPos));
+ outputColumns.add(info);
+ reversedExprs.put(outputColumnNames.get(outputPos), tag);
+ outputPos++;
+ }
+
+ exprMap.put(tag, new ArrayList<ExprNodeDesc>(descriptors.values()));
+ colExprMap.putAll(descriptors);
+ childOps[pos] = inputRS;
+ }
+
+ boolean noOuterJoin = joinType != JoinType.FULLOUTER
+ && joinType != JoinType.LEFTOUTER
+ && joinType != JoinType.RIGHTOUTER;
+ JoinDesc desc = new JoinDesc(exprMap, outputColumnNames, noOuterJoin,
+ joinCondns, joinKeys);
+ desc.setReversedExprs(reversedExprs);
+
+ JoinOperator joinOp = (JoinOperator) OperatorFactory.getAndMakeChild(desc,
+ new RowSchema(outputColumns), childOps);
+ joinOp.setColumnExprMap(colExprMap);
+ joinOp.setPosToAliasMap(posToAliasMap);
+
+ // TODO: null safes?
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Generated " + joinOp + " with row schema: [" + joinOp.getSchema() + "]");
+ }
+
+ return joinOp;
+ }
+
+ private static JoinType extractJoinType(HiveJoin join) {
+ // UNIQUE
+ if (join.isDistinct()) {
+ return JoinType.UNIQUE;
+ }
+ // SEMIJOIN
+ if (join.isLeftSemiJoin()) {
+ return JoinType.LEFTSEMI;
+ }
+ // OUTER AND INNER JOINS
+ JoinType resultJoinType;
+ switch (join.getJoinType()) {
+ case FULL:
+ resultJoinType = JoinType.FULLOUTER;
+ break;
+ case LEFT:
+ resultJoinType = JoinType.LEFTOUTER;
+ break;
+ case RIGHT:
+ resultJoinType = JoinType.RIGHTOUTER;
+ break;
+ default:
+ resultJoinType = JoinType.INNER;
+ break;
+ }
+ return resultJoinType;
+ }
+
+ private static Map<String, ExprNodeDesc> buildBacktrackFromReduceSink(
+ ReduceSinkOperator rsOp, Operator<?> inputOp) {
+ return buildBacktrackFromReduceSink(0, inputOp.getSchema().getColumnNames(),
+ rsOp.getConf().getOutputKeyColumnNames(), rsOp.getConf().getOutputValueColumnNames(),
+ rsOp.getValueIndex(), inputOp);
+ }
+
+ private static Map<String, ExprNodeDesc> buildBacktrackFromReduceSink(int initialPos,
+ List<String> outputColumnNames, List<String> keyColNames, List<String> valueColNames,
+ int[] index, Operator<?> inputOp) {
+ Map<String, ExprNodeDesc> columnDescriptors = new LinkedHashMap<String, ExprNodeDesc>();
+ for (int i = 0; i < index.length; i++) {
+ ColumnInfo info = new ColumnInfo(inputOp.getSchema().getSignature().get(i));
+ String field;
+ if (index[i] >= 0) {
+ field = Utilities.ReduceField.KEY + "." + keyColNames.get(index[i]);
+ } else {
+ field = Utilities.ReduceField.VALUE + "." + valueColNames.get(-index[i] - 1);
+ }
+ ExprNodeColumnDesc desc = new ExprNodeColumnDesc(info.getType(),
+ field, info.getTabAlias(), info.getIsVirtualCol());
+ columnDescriptors.put(outputColumnNames.get(initialPos+i), desc);
+ }
+ return columnDescriptors;
+ }
+
+ private static ExprNodeDesc convertToExprNode(RexNode rn, RelNode inputRel,
+ String tabAlias) {
+ return (ExprNodeDesc) rn.accept(new ExprNodeConverter(tabAlias,
+ inputRel.getRowType(), false));
+ }
+
+ private static ArrayList<ColumnInfo> createColInfos(Operator<?> input) {
+ ArrayList<ColumnInfo> cInfoLst = new ArrayList<ColumnInfo>();
+ for (ColumnInfo ci : input.getSchema().getSignature()) {
+ cInfoLst.add(new ColumnInfo(ci));
+ }
+ return cInfoLst;
+ }
+
+ private static Pair<ArrayList<ColumnInfo>, Map<Integer, VirtualColumn>> createColInfos(
+ List<RexNode> calciteExprs, List<ExprNodeDesc> hiveExprs, List<String> projNames,
+ OpAttr inpOpAf) {
+ if (hiveExprs.size() != projNames.size()) {
+ throw new RuntimeException("Column expressions list doesn't match Column Names list");
+ }
+
+ RexNode rexN;
+ ExprNodeDesc pe;
+ ArrayList<ColumnInfo> colInfos = new ArrayList<ColumnInfo>();
+ VirtualColumn vc;
+ Map<Integer, VirtualColumn> newVColMap = new HashMap<Integer, VirtualColumn>();
+ for (int i = 0; i < hiveExprs.size(); i++) {
+ pe = hiveExprs.get(i);
+ rexN = calciteExprs.get(i);
+ vc = null;
+ if (rexN instanceof RexInputRef) {
+ vc = inpOpAf.vcolMap.get(((RexInputRef) rexN).getIndex());
+ if (vc != null) {
+ newVColMap.put(i, vc);
+ }
+ }
+ colInfos
+ .add(new ColumnInfo(projNames.get(i), pe.getTypeInfo(), inpOpAf.tabAlias, vc != null));
+ }
+
+ return new Pair<ArrayList<ColumnInfo>, Map<Integer, VirtualColumn>>(colInfos, newVColMap);
+ }
+
+}
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/CalcitePlanner.java Thu Feb 19 18:41:04 2015
@@ -132,8 +132,10 @@ import org.apache.hadoop.hive.ql.optimiz
import org.apache.hadoop.hive.ql.optimizer.calcite.rules.HiveFilterJoinRule;
import org.apache.hadoop.hive.ql.optimizer.calcite.rules.HiveFilterProjectTransposeRule;
import org.apache.hadoop.hive.ql.optimizer.calcite.rules.HiveFilterSetOpTransposeRule;
+import org.apache.hadoop.hive.ql.optimizer.calcite.rules.HiveJoinAddNotNullRule;
import org.apache.hadoop.hive.ql.optimizer.calcite.rules.HivePartitionPruneRule;
import org.apache.hadoop.hive.ql.optimizer.calcite.translator.ASTConverter;
+import org.apache.hadoop.hive.ql.optimizer.calcite.translator.HiveOpConverter;
import org.apache.hadoop.hive.ql.optimizer.calcite.translator.JoinCondTypeCheckProcFactory;
import org.apache.hadoop.hive.ql.optimizer.calcite.translator.JoinTypeCheckCtx;
import org.apache.hadoop.hive.ql.optimizer.calcite.translator.RexNodeConverter;
@@ -215,39 +217,43 @@ public class CalcitePlanner extends Sema
if (cboCtx.type == PreCboCtx.Type.CTAS) {
queryForCbo = cboCtx.nodeOfInterest; // nodeOfInterest is the query
}
- runCBO = canHandleAstForCbo(queryForCbo, getQB(), cboCtx);
+ runCBO = canCBOHandleAst(queryForCbo, getQB(), cboCtx);
if (runCBO) {
disableJoinMerge = true;
boolean reAnalyzeAST = false;
try {
- // 1. Gen Optimized AST
- ASTNode newAST = getOptimizedAST();
+ if (this.conf.getBoolVar(HiveConf.ConfVars.HIVE_CBO_RETPATH_HIVEOP)) {
+ sinkOp = getOptimizedHiveOPDag();
+ } else {
+ // 1. Gen Optimized AST
+ ASTNode newAST = getOptimizedAST();
- // 1.1. Fix up the query for insert/ctas
- newAST = fixUpCtasAndInsertAfterCbo(ast, newAST, cboCtx);
+ // 1.1. Fix up the query for insert/ctas
+ newAST = fixUpCtasAndInsertAfterCbo(ast, newAST, cboCtx);
- // 2. Regen OP plan from optimized AST
- init(false);
- if (cboCtx.type == PreCboCtx.Type.CTAS) {
- // Redo create-table analysis, because it's not part of doPhase1.
- setAST(newAST);
- newAST = reAnalyzeCtasAfterCbo(newAST);
- }
- Phase1Ctx ctx_1 = initPhase1Ctx();
- if (!doPhase1(newAST, getQB(), ctx_1, null)) {
- throw new RuntimeException("Couldn't do phase1 on CBO optimized query plan");
+ // 2. Regen OP plan from optimized AST
+ init(false);
+ if (cboCtx.type == PreCboCtx.Type.CTAS) {
+ // Redo create-table analysis, because it's not part of doPhase1.
+ setAST(newAST);
+ newAST = reAnalyzeCtasAfterCbo(newAST);
+ }
+ Phase1Ctx ctx_1 = initPhase1Ctx();
+ if (!doPhase1(newAST, getQB(), ctx_1, null)) {
+ throw new RuntimeException("Couldn't do phase1 on CBO optimized query plan");
+ }
+ // unfortunately making prunedPartitions immutable is not possible
+ // here with SemiJoins not all tables are costed in CBO, so their
+ // PartitionList is not evaluated until the run phase.
+ getMetaData(getQB());
+
+ disableJoinMerge = false;
+ sinkOp = genPlan(getQB());
+ LOG.info("CBO Succeeded; optimized logical plan.");
+ LOG.debug(newAST.dump());
}
- // unfortunately making prunedPartitions immutable is not possible
- // here with SemiJoins not all tables are costed in CBO, so their
- // PartitionList is not evaluated until the run phase.
- getMetaData(getQB());
-
- disableJoinMerge = false;
- sinkOp = genPlan(getQB());
- LOG.info("CBO Succeeded; optimized logical plan.");
- LOG.debug(newAST.dump());
} catch (Exception e) {
boolean isMissingStats = noColsMissingStats.get() > 0;
if (isMissingStats) {
@@ -304,7 +310,7 @@ public class CalcitePlanner extends Sema
* If top level QB is query then everything below it must also be
* Query.
*/
- boolean canHandleAstForCbo(ASTNode ast, QB qb, PreCboCtx cboCtx) {
+ boolean canCBOHandleAst(ASTNode ast, QB qb, PreCboCtx cboCtx) {
int root = ast.getToken().getType();
boolean needToLogMessage = STATIC_LOG.isInfoEnabled();
boolean isSupportedRoot = root == HiveParser.TOK_QUERY || root == HiveParser.TOK_EXPLAIN
@@ -578,6 +584,31 @@ public class CalcitePlanner extends Sema
return optiqOptimizedAST;
}
+ /**
+ * Get Optimized Hive Operator DAG for the given QB tree in the semAnalyzer.
+ *
+ * @return Optimized Hive operator tree
+ * @throws SemanticException
+ */
+ Operator getOptimizedHiveOPDag() throws SemanticException {
+ RelNode optimizedOptiqPlan = null;
+ CalcitePlannerAction calcitePlannerAction = new CalcitePlannerAction(prunedPartitions);
+
+ try {
+ optimizedOptiqPlan = Frameworks.withPlanner(calcitePlannerAction, Frameworks
+ .newConfigBuilder().typeSystem(new HiveTypeSystemImpl()).build());
+ } catch (Exception e) {
+ rethrowCalciteException(e);
+ throw new AssertionError("rethrowCalciteException didn't throw for " + e.getMessage());
+ }
+
+ Operator hiveRoot = new HiveOpConverter(topOps, HiveOpConverter.getAggOPMode(conf),
+ conf.getVar(HiveConf.ConfVars.HIVEMAPREDMODE).equalsIgnoreCase("strict")).convert(optimizedOptiqPlan);
+ RowResolver hiveRootRR = genRowResolver(hiveRoot, getQB());
+ opParseCtx.put(hiveRoot, new OpParseContext(hiveRootRR));
+ return genFileSinkPlan(getQB().getParseInfo().getClauseNames().iterator().next(), getQB(), hiveRoot);
+ }
+
/***
* Unwraps Calcite Invocation exceptions coming meta data provider chain and
* obtains the real cause.
@@ -654,6 +685,24 @@ public class CalcitePlanner extends Sema
|| t instanceof UndeclaredThrowableException;
}
+ private RowResolver genRowResolver(Operator op, QB qb) {
+ RowResolver rr = new RowResolver();
+ String subqAlias = (qb.getAliases().size() == 1 && qb.getSubqAliases().size() == 1) ? qb
+ .getAliases().get(0) : null;
+
+ for (ColumnInfo ci : op.getSchema().getSignature()) {
+ try {
+ rr.putWithCheck((subqAlias != null) ? subqAlias : ci.getTabAlias(),
+ ci.getAlias() != null ? ci.getAlias() : ci.getInternalName(), ci.getInternalName(),
+ new ColumnInfo(ci));
+ } catch (SemanticException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return rr;
+ }
+
/**
* Code responsible for Calcite plan generation and optimization.
*/
@@ -773,6 +822,7 @@ public class CalcitePlanner extends Sema
ReduceExpressionsRule.PROJECT_INSTANCE,
ReduceExpressionsRule.FILTER_INSTANCE,
ReduceExpressionsRule.JOIN_INSTANCE,
+ HiveJoinAddNotNullRule.INSTANCE,
new HiveFilterProjectTransposeRule(
Filter.class, HiveFilter.DEFAULT_FILTER_FACTORY, HiveProject.class,
HiveProject.DEFAULT_PROJECT_FACTORY), new HiveFilterSetOpTransposeRule(
@@ -1167,7 +1217,7 @@ public class CalcitePlanner extends Sema
}
// 2. Get Table Metadata
- Table tab = qb.getMetaData().getSrcForAlias(tableAlias);
+ Table tabMetaData = qb.getMetaData().getSrcForAlias(tableAlias);
// 3. Get Table Logical Schema (Row Type)
// NOTE: Table logical schema = Non Partition Cols + Partition Cols +
@@ -1175,7 +1225,7 @@ public class CalcitePlanner extends Sema
// 3.1 Add Column info for non partion cols (Object Inspector fields)
@SuppressWarnings("deprecation")
- StructObjectInspector rowObjectInspector = (StructObjectInspector) tab.getDeserializer()
+ StructObjectInspector rowObjectInspector = (StructObjectInspector) tabMetaData.getDeserializer()
.getObjectInspector();
List<? extends StructField> fields = rowObjectInspector.getAllStructFieldRefs();
ColumnInfo colInfo;
@@ -1197,7 +1247,7 @@ public class CalcitePlanner extends Sema
ArrayList<ColumnInfo> partitionColumns = new ArrayList<ColumnInfo>();
// 3.2 Add column info corresponding to partition columns
- for (FieldSchema part_col : tab.getPartCols()) {
+ for (FieldSchema part_col : tabMetaData.getPartCols()) {
colName = part_col.getName();
colInfo = new ColumnInfo(colName,
TypeInfoFactory.getPrimitiveTypeInfo(part_col.getType()), tableAlias, true);
@@ -1207,6 +1257,7 @@ public class CalcitePlanner extends Sema
}
// 3.3 Add column info corresponding to virtual columns
+ List<VirtualColumn> virtualCols = new ArrayList<VirtualColumn>();
Iterator<VirtualColumn> vcs = VirtualColumn.getRegistry(conf).iterator();
while (vcs.hasNext()) {
VirtualColumn vc = vcs.next();
@@ -1214,24 +1265,27 @@ public class CalcitePlanner extends Sema
vc.getIsHidden());
rr.put(tableAlias, vc.getName(), colInfo);
cInfoLst.add(colInfo);
+ virtualCols.add(vc);
}
// 3.4 Build row type from field <type, name>
RelDataType rowType = TypeConverter.getType(cluster, rr, null);
// 4. Build RelOptAbstractTable
- String fullyQualifiedTabName = tab.getDbName();
- if (fullyQualifiedTabName != null && !fullyQualifiedTabName.isEmpty())
- fullyQualifiedTabName = fullyQualifiedTabName + "." + tab.getTableName();
- else
- fullyQualifiedTabName = tab.getTableName();
+ String fullyQualifiedTabName = tabMetaData.getDbName();
+ if (fullyQualifiedTabName != null && !fullyQualifiedTabName.isEmpty()) {
+ fullyQualifiedTabName = fullyQualifiedTabName + "." + tabMetaData.getTableName()
+ + "." + tableAlias;
+ }
+ else {
+ fullyQualifiedTabName = tabMetaData.getTableName() + "." + tableAlias;
+ }
RelOptHiveTable optTable = new RelOptHiveTable(relOptSchema, fullyQualifiedTabName,
- tableAlias, rowType, tab, nonPartitionColumns, partitionColumns, conf, partitionCache,
- noColsMissingStats);
+ tableAlias, rowType, tabMetaData, nonPartitionColumns, partitionColumns, virtualCols, conf,
+ partitionCache, noColsMissingStats, getAliasId(tableAlias, qb));
// 5. Build Hive Table Scan Rel
- tableRel = new HiveTableScan(cluster, cluster.traitSetOf(HiveRelNode.CONVENTION), optTable,
- rowType);
+ tableRel = new HiveTableScan(cluster, cluster.traitSetOf(HiveRelNode.CONVENTION), optTable);
// 6. Add Schema(RR) to RelNode-Schema map
ImmutableMap<String, Integer> hiveToCalciteColMap = buildHiveToCalciteColumnMap(rr,
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java Thu Feb 19 18:41:04 2015
@@ -228,9 +228,9 @@ public class SemanticAnalyzer extends Ba
private HashMap<TableScanOperator, ExprNodeDesc> opToPartPruner;
private HashMap<TableScanOperator, PrunedPartitionList> opToPartList;
- private HashMap<String, Operator<? extends OperatorDesc>> topOps;
- private final HashMap<String, Operator<? extends OperatorDesc>> topSelOps;
- private final LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtx;
+ protected HashMap<String, Operator<? extends OperatorDesc>> topOps;
+ private HashMap<String, Operator<? extends OperatorDesc>> topSelOps;
+ protected LinkedHashMap<Operator<? extends OperatorDesc>, OpParseContext> opParseCtx;
private List<LoadTableDesc> loadTableWork;
private List<LoadFileDesc> loadFileWork;
private final Map<JoinOperator, QBJoinTree> joinContext;
@@ -6123,7 +6123,7 @@ public class SemanticAnalyzer extends Ba
}
@SuppressWarnings("nls")
- private Operator genFileSinkPlan(String dest, QB qb, Operator input)
+ protected Operator genFileSinkPlan(String dest, QB qb, Operator input)
throws SemanticException {
RowResolver inputRR = opParseCtx.get(input).getRowResolver();
@@ -9226,7 +9226,7 @@ public class SemanticAnalyzer extends Ba
return equalsExpr;
}
- private String getAliasId(String alias, QB qb) {
+ protected String getAliasId(String alias, QB qb) {
return (qb.getId() == null ? alias : qb.getId() + ":" + alias).toLowerCase();
}
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/JoinDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/JoinDesc.java?rev=1660968&r1=1660967&r2=1660968&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/JoinDesc.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/JoinDesc.java Thu Feb 19 18:41:04 2015
@@ -107,6 +107,13 @@ public class JoinDesc extends AbstractOp
}
public JoinDesc(final Map<Byte, List<ExprNodeDesc>> exprs,
+ List<String> outputColumnNames, final boolean noOuterJoin,
+ final JoinCondDesc[] conds, ExprNodeDesc[][] joinKeys) {
+ this (exprs, outputColumnNames, noOuterJoin, conds,
+ new HashMap<Byte, List<ExprNodeDesc>>(), joinKeys);
+ }
+
+ public JoinDesc(final Map<Byte, List<ExprNodeDesc>> exprs,
List<String> outputColumnNames, final boolean noOuterJoin,
final JoinCondDesc[] conds, final Map<Byte, List<ExprNodeDesc>> filters,
ExprNodeDesc[][] joinKeys) {