You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by gu...@apache.org on 2014/07/01 04:15:24 UTC
svn commit: r1606964 [1/2] - in /hive/branches/cbo: ./
ql/src/java/org/apache/hadoop/hive/ql/
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/
ql/src/java/org/apache/hadoop/hive/ql/io/ ql/src/java/org/apach...
Author: gunther
Date: Tue Jul 1 02:15:23 2014
New Revision: 1606964
URL: http://svn.apache.org/r1606964
Log:
Merge latest trunk into branch. (Gunther Hagleitner)
Added:
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/PTFTopNHash.java
- copied unchanged from r1606962, hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/PTFTopNHash.java
hive/branches/cbo/ql/src/test/queries/clientpositive/columnstats_partlvl_dp.q
- copied unchanged from r1606962, hive/trunk/ql/src/test/queries/clientpositive/columnstats_partlvl_dp.q
hive/branches/cbo/ql/src/test/queries/clientpositive/groupby_duplicate_key.q
- copied unchanged from r1606962, hive/trunk/ql/src/test/queries/clientpositive/groupby_duplicate_key.q
hive/branches/cbo/ql/src/test/queries/clientpositive/windowing_streaming.q
- copied unchanged from r1606962, hive/trunk/ql/src/test/queries/clientpositive/windowing_streaming.q
hive/branches/cbo/ql/src/test/results/clientpositive/columnstats_partlvl_dp.q.out
- copied unchanged from r1606962, hive/trunk/ql/src/test/results/clientpositive/columnstats_partlvl_dp.q.out
hive/branches/cbo/ql/src/test/results/clientpositive/groupby_duplicate_key.q.out
- copied unchanged from r1606962, hive/trunk/ql/src/test/results/clientpositive/groupby_duplicate_key.q.out
hive/branches/cbo/ql/src/test/results/clientpositive/windowing_streaming.q.out
- copied unchanged from r1606962, hive/trunk/ql/src/test/results/clientpositive/windowing_streaming.q.out
Removed:
hive/branches/cbo/ql/src/test/queries/clientnegative/columnstats_partlvl_dp.q
hive/branches/cbo/ql/src/test/queries/clientnegative/columnstats_partlvl_incorrect_num_keys.q
hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_dp.q.out
hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_incorrect_num_keys.q.out
Modified:
hive/branches/cbo/ (props changed)
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsTask.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/TopNHash.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorReduceSinkOperator.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.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/parse/TaskCompiler.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ColumnStatsDesc.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/PlanUtils.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ptf/WindowTableFunctionDef.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpProcFactory.java
hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/udf/ptf/WindowingTableFunction.java
hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_invalid_values.q.out
hive/branches/cbo/ql/src/test/results/clientpositive/columnstats_partlvl.q.out
hive/branches/cbo/ql/src/test/results/compiler/plan/groupby1.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/groupby4.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/groupby5.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/groupby6.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join1.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join2.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join3.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join4.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join5.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join6.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join7.q.xml
hive/branches/cbo/ql/src/test/results/compiler/plan/join8.q.xml
Propchange: hive/branches/cbo/
------------------------------------------------------------------------------
Merged /hive/trunk:r1606273-1606962
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java Tue Jul 1 02:15:23 2014
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.ql;
import org.antlr.runtime.tree.Tree;
+import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.metadata.HiveUtils;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.ASTNodeOrigin;
@@ -29,12 +30,6 @@ import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import org.antlr.runtime.tree.Tree;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.metadata.HiveUtils;
-import org.apache.hadoop.hive.ql.parse.ASTNode;
-import org.apache.hadoop.hive.ql.parse.ASTNodeOrigin;
-
/**
* List of all error messages.
* This list contains both compile time and run-time errors.
@@ -80,7 +75,6 @@ public enum ErrorMsg {
INVALID_JOIN_CONDITION_2(10018, "Neither left nor right aliases encountered in JOIN"),
INVALID_JOIN_CONDITION_3(10019, "OR not supported in JOIN currently"),
INVALID_TRANSFORM(10020, "TRANSFORM with other SELECT columns not supported"),
- DUPLICATE_GROUPBY_KEY(10021, "Repeated key in GROUP BY"),
UNSUPPORTED_MULTIPLE_DISTINCTS(10022, "DISTINCT on different columns not supported" +
" with skew in data"),
NO_SUBQUERY_ALIAS(10023, "No alias for subquery"),
@@ -444,12 +438,8 @@ public enum ErrorMsg {
COLUMNSTATSCOLLECTOR_INVALID_PART_KEY(30005, "Invalid partitioning key specified in ANALYZE " +
"statement"),
- COLUMNSTATSCOLLECTOR_INCORRECT_NUM_PART_KEY(30006, "Incorrect number of partitioning key " +
- "specified in ANALYZE statement"),
COLUMNSTATSCOLLECTOR_INVALID_PARTITION(30007, "Invalid partitioning key/value specified in " +
"ANALYZE statement"),
- COLUMNSTATSCOLLECTOR_INVALID_SYNTAX(30008, "Dynamic partitioning is not supported yet while " +
- "gathering column statistics through ANALYZE statement"),
COLUMNSTATSCOLLECTOR_PARSE_ERROR(30009, "Encountered parse error while parsing rewritten query"),
COLUMNSTATSCOLLECTOR_IO_ERROR(30010, "Encountered I/O exception while parsing rewritten query"),
DROP_COMMAND_NOT_ALLOWED_FOR_PARTITION(30011, "Partition protected from being dropped"),
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsTask.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsTask.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsTask.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnStatsTask.java Tue Jul 1 02:15:23 2014
@@ -28,6 +28,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.common.type.HiveDecimal;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.Warehouse;
import org.apache.hadoop.hive.metastore.api.BinaryColumnStatsData;
import org.apache.hadoop.hive.metastore.api.BooleanColumnStatsData;
import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
@@ -37,12 +38,14 @@ import org.apache.hadoop.hive.metastore.
import org.apache.hadoop.hive.metastore.api.Decimal;
import org.apache.hadoop.hive.metastore.api.DecimalColumnStatsData;
import org.apache.hadoop.hive.metastore.api.DoubleColumnStatsData;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.LongColumnStatsData;
+import org.apache.hadoop.hive.metastore.api.MetaException;
import org.apache.hadoop.hive.metastore.api.StringColumnStatsData;
-import org.apache.hadoop.hive.ql.CommandNeedRetryException;
import org.apache.hadoop.hive.ql.DriverContext;
import org.apache.hadoop.hive.ql.QueryPlan;
import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.plan.ColumnStatsWork;
import org.apache.hadoop.hive.ql.plan.api.StageType;
import org.apache.hadoop.hive.ql.session.SessionState;
@@ -65,8 +68,6 @@ import org.apache.hadoop.util.StringUtil
public class ColumnStatsTask extends Task<ColumnStatsWork> implements Serializable {
private static final long serialVersionUID = 1L;
private FetchOperator ftOp;
- private int totalRows;
- private int numRows = 0;
private static transient final Log LOG = LogFactory.getLog(ColumnStatsTask.class);
public ColumnStatsTask() {
@@ -262,11 +263,7 @@ public class ColumnStatsTask extends Tas
}
}
- private ColumnStatistics constructColumnStatsFromPackedRow(ObjectInspector oi,
- Object o) throws HiveException {
- if (oi.getCategory() != ObjectInspector.Category.STRUCT) {
- throw new HiveException("Unexpected object type encountered while unpacking row");
- }
+ private List<ColumnStatistics> constructColumnStatsFromPackedRows() throws HiveException, MetaException, IOException {
String dbName = SessionState.get().getCurrentDatabase();
String tableName = work.getColStats().getTableName();
@@ -275,33 +272,51 @@ public class ColumnStatsTask extends Tas
List<String> colType = work.getColStats().getColType();
boolean isTblLevel = work.getColStats().isTblLevel();
- if (!isTblLevel) {
- partName = work.getColStats().getPartName();
- }
+ List<ColumnStatistics> stats = new ArrayList<ColumnStatistics>();
+ InspectableObject packedRow;
+ while ((packedRow = ftOp.getNextRow()) != null) {
+ if (packedRow.oi.getCategory() != ObjectInspector.Category.STRUCT) {
+ throw new HiveException("Unexpected object type encountered while unpacking row");
+ }
- ColumnStatisticsDesc statsDesc = getColumnStatsDesc(dbName, tableName, partName, isTblLevel);
+ List<ColumnStatisticsObj> statsObjs = new ArrayList<ColumnStatisticsObj>();
+ StructObjectInspector soi = (StructObjectInspector) packedRow.oi;
+ List<? extends StructField> fields = soi.getAllStructFieldRefs();
+ List<Object> list = soi.getStructFieldsDataAsList(packedRow.o);
+
+ Table tbl = db.getTable(dbName,tableName);
+ List<FieldSchema> partColSchema = tbl.getPartCols();
+ // Partition columns are appended at end, we only care about stats column
+ for (int i = 0; i < fields.size() - partColSchema.size(); i++) {
+ // Get the field objectInspector, fieldName and the field object.
+ ObjectInspector foi = fields.get(i).getFieldObjectInspector();
+ Object f = (list == null ? null : list.get(i));
+ String fieldName = fields.get(i).getFieldName();
+ ColumnStatisticsObj statsObj = new ColumnStatisticsObj();
+ statsObj.setColName(colName.get(i));
+ statsObj.setColType(colType.get(i));
+ unpackStructObject(foi, f, fieldName, statsObj);
+ statsObjs.add(statsObj);
+ }
- List<ColumnStatisticsObj> statsObjs = new ArrayList<ColumnStatisticsObj>();
- StructObjectInspector soi = (StructObjectInspector) oi;
- List<? extends StructField> fields = soi.getAllStructFieldRefs();
- List<Object> list = soi.getStructFieldsDataAsList(o);
+ if (!isTblLevel) {
+ List<String> partVals = new ArrayList<String>();
+ // Iterate over partition columns to figure out partition name
+ for (int i = fields.size() - partColSchema.size(); i < fields.size(); i++) {
+ partVals.add(((PrimitiveObjectInspector)fields.get(i).getFieldObjectInspector()).
+ getPrimitiveJavaObject(list.get(i)).toString());
+ }
+ partName = Warehouse.makePartName(partColSchema, partVals);
+ }
- for (int i = 0; i < fields.size(); i++) {
- // Get the field objectInspector, fieldName and the field object.
- ObjectInspector foi = fields.get(i).getFieldObjectInspector();
- Object f = (list == null ? null : list.get(i));
- String fieldName = fields.get(i).getFieldName();
- ColumnStatisticsObj statsObj = new ColumnStatisticsObj();
- statsObj.setColName(colName.get(i));
- statsObj.setColType(colType.get(i));
- unpackStructObject(foi, f, fieldName, statsObj);
- statsObjs.add(statsObj);
+ ColumnStatisticsDesc statsDesc = getColumnStatsDesc(dbName, tableName, partName, isTblLevel);
+ ColumnStatistics colStats = new ColumnStatistics();
+ colStats.setStatsDesc(statsDesc);
+ colStats.setStatsObj(statsObjs);
+ stats.add(colStats);
}
-
- ColumnStatistics colStats = new ColumnStatistics();
- colStats.setStatsDesc(statsDesc);
- colStats.setStatsObj(statsObjs);
- return colStats;
+ ftOp.clearFetchContext();
+ return stats;
}
private ColumnStatisticsDesc getColumnStatsDesc(String dbName, String tableName,
@@ -320,53 +335,24 @@ public class ColumnStatsTask extends Tas
return statsDesc;
}
- private int persistPartitionStats() throws HiveException {
- InspectableObject io = null;
- // Fetch result of the analyze table .. compute statistics for columns ..
- try {
- io = fetchColumnStats();
- } catch (IOException e) {
- e.printStackTrace();
- } catch (CommandNeedRetryException e) {
- e.printStackTrace();
- }
-
- if (io != null) {
- // Construct a column statistics object from the result
- ColumnStatistics colStats = constructColumnStatsFromPackedRow(io.oi, io.o);
-
- // Persist the column statistics object to the metastore
- try {
- db.updatePartitionColumnStatistics(colStats);
- } catch (Exception e) {
- e.printStackTrace();
- }
+ private int persistPartitionStats() throws HiveException, MetaException, IOException {
+
+ // Fetch result of the analyze table partition (p1=c1).. compute statistics for columns ..
+ // Construct a column statistics object from the result
+ List<ColumnStatistics> colStats = constructColumnStatsFromPackedRows();
+ // Persist the column statistics object to the metastore
+ for (ColumnStatistics colStat : colStats) {
+ db.updatePartitionColumnStatistics(colStat);
}
return 0;
}
- private int persistTableStats() throws HiveException {
- InspectableObject io = null;
+ private int persistTableStats() throws HiveException, MetaException, IOException {
// Fetch result of the analyze table .. compute statistics for columns ..
- try {
- io = fetchColumnStats();
- } catch (IOException e) {
- e.printStackTrace();
- } catch (CommandNeedRetryException e) {
- e.printStackTrace();
- }
-
- if (io != null) {
- // Construct a column statistics object from the result
- ColumnStatistics colStats = constructColumnStatsFromPackedRow(io.oi, io.o);
-
- // Persist the column statistics object to the metastore
- try {
- db.updateTableColumnStatistics(colStats);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
+ // Construct a column statistics object from the result
+ ColumnStatistics colStats = constructColumnStatsFromPackedRows().get(0);
+ // Persist the column statistics object to the metastore
+ db.updateTableColumnStatistics(colStats);
return 0;
}
@@ -379,42 +365,11 @@ public class ColumnStatsTask extends Tas
return persistPartitionStats();
}
} catch (Exception e) {
- e.printStackTrace();
+ LOG.info(e);
}
return 1;
}
- private InspectableObject fetchColumnStats() throws IOException, CommandNeedRetryException {
- InspectableObject io = null;
-
- try {
- int rowsRet = work.getLeastNumRows();
- if (rowsRet <= 0) {
- rowsRet = ColumnStatsWork.getLimit() >= 0 ?
- Math.min(ColumnStatsWork.getLimit() - totalRows, 1) : 1;
- }
- if (rowsRet <= 0) {
- ftOp.clearFetchContext();
- return null;
- }
- while (numRows < rowsRet) {
- if ((io = ftOp.getNextRow()) == null) {
- if (work.getLeastNumRows() > 0) {
- throw new CommandNeedRetryException();
- }
- }
- numRows++;
- }
- return io;
- } catch (CommandNeedRetryException e) {
- throw e;
- } catch (IOException e) {
- throw e;
- } catch (Exception e) {
- throw new IOException(e);
- }
- }
-
@Override
public StageType getType() {
return StageType.COLUMNSTATS;
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/ReduceSinkOperator.java Tue Jul 1 02:15:23 2014
@@ -118,7 +118,6 @@ public class ReduceSinkOperator extends
protected transient TopNHash reducerHash = new TopNHash();
@Override
protected void initializeOp(Configuration hconf) throws HiveException {
-
try {
List<ExprNodeDesc> keys = conf.getKeyCols();
keyEval = new ExprNodeEvaluator[keys.size()];
@@ -174,7 +173,9 @@ public class ReduceSinkOperator extends
int limit = conf.getTopN();
float memUsage = conf.getTopNMemoryUsage();
+
if (limit >= 0 && memUsage > 0) {
+ reducerHash = conf.isPTFReduceSink() ? new PTFTopNHash() : reducerHash;
reducerHash.initialize(limit, memUsage, conf.isMapGroupBy(), this);
}
@@ -311,11 +312,17 @@ public class ReduceSinkOperator extends
} else {
hashCode = computeHashCode(row, buckNum);
}
-
+
firstKey.setHashCode(hashCode);
+ /*
+ * in case of TopN for windowing, we need to distinguish between rows with
+ * null partition keys and rows with value 0 for partition keys.
+ */
+ boolean partKeyNull = conf.isPTFReduceSink() && partitionKeysAreNull(row);
+
// Try to store the first key. If it's not excluded, we will proceed.
- int firstIndex = reducerHash.tryStoreKey(firstKey);
+ int firstIndex = reducerHash.tryStoreKey(firstKey, partKeyNull);
if (firstIndex == TopNHash.EXCLUDE) return; // Nothing to do.
// Compute value and hashcode - we'd either store or forward them.
BytesWritable value = makeValueWritable(row);
@@ -324,7 +331,7 @@ public class ReduceSinkOperator extends
collect(firstKey, value);
} else {
assert firstIndex >= 0;
- reducerHash.storeValue(firstIndex, value, false);
+ reducerHash.storeValue(firstIndex, firstKey.hashCode(), value, false);
}
// All other distinct keys will just be forwarded. This could be optimized...
@@ -408,6 +415,19 @@ public class ReduceSinkOperator extends
return buckNum < 0 ? keyHashCode : keyHashCode * 31 + buckNum;
}
+ private boolean partitionKeysAreNull(Object row) throws HiveException {
+ if ( partitionEval.length != 0 ) {
+ for (int i = 0; i < partitionEval.length; i++) {
+ Object o = partitionEval[i].evaluate(row);
+ if ( o != null ) {
+ return false;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
// Serialize the keys and append the tag
protected HiveKey toHiveKey(Object obj, int tag, Integer distLength) throws SerDeException {
BinaryComparable key = (BinaryComparable)keySerializer.serialize(obj, keyObjectInspector);
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/TopNHash.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/TopNHash.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/TopNHash.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/TopNHash.java Tue Jul 1 02:15:23 2014
@@ -57,11 +57,11 @@ public class TopNHash {
public static final int EXCLUDE = -2; // Discard the row.
private static final int MAY_FORWARD = -3; // Vectorized - may forward the row, not sure yet.
- private BinaryCollector collector;
- private int topN;
+ protected BinaryCollector collector;
+ protected int topN;
- private long threshold; // max heap size
- private long usage;
+ protected long threshold; // max heap size
+ protected long usage;
// binary keys, values and hashCodes of rows, lined up by index
private byte[][] keys;
@@ -76,10 +76,10 @@ public class TopNHash {
// temporary single-batch context used for vectorization
private int batchNumForwards = 0; // whether current batch has any forwarded keys
private int[] indexToBatchIndex; // mapping of index (lined up w/keys) to index in the batch
- private int[] batchIndexToResult; // mapping of index in the batch (linear) to hash result
- private int batchSize; // Size of the current batch.
+ protected int[] batchIndexToResult; // mapping of index in the batch (linear) to hash result
+ protected int batchSize; // Size of the current batch.
- private boolean isEnabled = false;
+ protected boolean isEnabled = false;
private final Comparator<Integer> C = new Comparator<Integer>() {
public int compare(Integer o1, Integer o2) {
@@ -124,7 +124,7 @@ public class TopNHash {
* TopNHash.EXCLUDED if the row should be discarded;
* any other number if the row is to be stored; the index should be passed to storeValue.
*/
- public int tryStoreKey(HiveKey key) throws HiveException, IOException {
+ public int tryStoreKey(HiveKey key, boolean partColsIsNull) throws HiveException, IOException {
if (!isEnabled) {
return FORWARD; // short-circuit quickly - forward all rows
}
@@ -191,7 +191,7 @@ public class TopNHash {
* @param key the key.
* @param batchIndex The index of the key in the vectorized batch (sequential, not .selected).
*/
- public void tryStoreVectorizedKey(HiveKey key, int batchIndex)
+ public void tryStoreVectorizedKey(HiveKey key, boolean partColsIsNull, int batchIndex)
throws HiveException, IOException {
// Assumption - batchIndex is increasing; startVectorizedBatch was called
int size = indexes.size();
@@ -201,6 +201,13 @@ public class TopNHash {
hashes[index] = key.hashCode();
Integer collisionIndex = indexes.store(index);
if (null != collisionIndex) {
+ /*
+ * since there is a collision index will be used for the next value
+ * so have the map point back to original index.
+ */
+ if ( indexes instanceof HashForGroup ) {
+ indexes.store(collisionIndex);
+ }
// forward conditional on the survival of the corresponding key currently in indexes.
++batchNumForwards;
batchIndexToResult[batchIndex] = MAY_FORWARD - collisionIndex;
@@ -283,11 +290,12 @@ public class TopNHash {
/**
* Stores the value for the key in the heap.
* @param index The index, either from tryStoreKey or from tryStoreVectorizedKey result.
+ * @param hasCode hashCode of key, used by ptfTopNHash.
* @param value The value to store.
* @param keyHash The key hash to store.
* @param vectorized Whether the result is coming from a vectorized batch.
*/
- public void storeValue(int index, BytesWritable value, boolean vectorized) {
+ public void storeValue(int index, int hashCode, BytesWritable value, boolean vectorized) {
values[index] = Arrays.copyOf(value.getBytes(), value.getLength());
// Vectorized doesn't adjust usage for the keys while processing the batch
usage += values[index].length + (vectorized ? keys[index].length : 0);
@@ -369,7 +377,7 @@ public class TopNHash {
}
excluded = 0;
}
-
+
private interface IndexStore {
int size();
/**
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorReduceSinkOperator.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorReduceSinkOperator.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorReduceSinkOperator.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/exec/vector/VectorReduceSinkOperator.java Tue Jul 1 02:15:23 2014
@@ -24,6 +24,7 @@ import java.util.Random;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.exec.PTFTopNHash;
import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.exec.TopNHash;
import org.apache.hadoop.hive.ql.exec.vector.expressions.VectorExpression;
@@ -208,6 +209,7 @@ public class VectorReduceSinkOperator ex
int limit = conf.getTopN();
float memUsage = conf.getTopNMemoryUsage();
if (limit >= 0 && memUsage > 0) {
+ reducerHash = conf.isPTFReduceSink() ? new PTFTopNHash() : reducerHash;
reducerHash.initialize(limit, memUsage, conf.isMapGroupBy(), this);
}
@@ -293,7 +295,12 @@ public class VectorReduceSinkOperator ex
firstKey.setHashCode(hashCode);
if (useTopN) {
- reducerHash.tryStoreVectorizedKey(firstKey, batchIndex);
+ /*
+ * in case of TopN for windowing, we need to distinguish between
+ * rows with null partition keys and rows with value 0 for partition keys.
+ */
+ boolean partkeysNull = conf.isPTFReduceSink() && partitionKeysAreNull(vrg, rowIndex);
+ reducerHash.tryStoreVectorizedKey(firstKey, partkeysNull, batchIndex);
} else {
// No TopN, just forward the first key and all others.
BytesWritable value = makeValueWritable(vrg, rowIndex);
@@ -322,9 +329,9 @@ public class VectorReduceSinkOperator ex
hashCode = firstKey.hashCode();
collect(firstKey, value);
} else {
- reducerHash.storeValue(result, value, true);
- distKeyLength = reducerHash.getVectorizedKeyDistLength(batchIndex);
hashCode = reducerHash.getVectorizedKeyHashCode(batchIndex);
+ reducerHash.storeValue(result, hashCode, value, true);
+ distKeyLength = reducerHash.getVectorizedKeyDistLength(batchIndex);
}
// Now forward other the rows if there's multi-distinct (but see TODO in forward...).
// Unfortunately, that means we will have to rebuild the cachedKeys. Start at 1.
@@ -445,6 +452,22 @@ public class VectorReduceSinkOperator ex
return buckNum < 0 ? keyHashCode : keyHashCode * 31 + buckNum;
}
+ private boolean partitionKeysAreNull(VectorizedRowBatch vrg, int rowIndex)
+ throws HiveException {
+ if (partitionEval.length != 0) {
+ for (int p = 0; p < partitionEval.length; p++) {
+ ColumnVector columnVector = vrg.cols[partitionEval[p].getOutputColumn()];
+ Object partitionValue = partitionWriters[p].writeValue(columnVector,
+ rowIndex);
+ if (partitionValue != null) {
+ return false;
+ }
+ }
+ return true;
+ }
+ return false;
+ }
+
private int computeBucketNumber(VectorizedRowBatch vrg, int rowIndex, int numBuckets) throws HiveException {
int bucketNum = 0;
for (int p = 0; p < bucketEval.length; p++) {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java Tue Jul 1 02:15:23 2014
@@ -59,6 +59,8 @@ import org.apache.hadoop.io.compress.Dec
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.ReflectionUtils;
+import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.*;
+
/**
* <code>RCFile</code>s, short of Record Columnar File, are flat files
* consisting of binary key/value pairs, which shares much similarity with
@@ -340,8 +342,15 @@ public class RCFile {
private static final Log LOG = LogFactory.getLog(RCFile.class);
+ // internal variable
public static final String COLUMN_NUMBER_METADATA_STR = "hive.io.rcfile.column.number";
+ public static final String RECORD_INTERVAL_CONF_STR = HIVE_RCFILE_RECORD_INTERVAL.varname;
+
+ public static final String COLUMN_NUMBER_CONF_STR = HIVE_RCFILE_COLUMN_NUMBER_CONF.varname;
+
+ public static final String TOLERATE_CORRUPTIONS_CONF_STR = HIVE_RCFILE_TOLERATE_CORRUPTIONS.varname;
+
// HACK: We actually need BlockMissingException, but that is not available
// in all hadoop versions.
public static final String BLOCK_MISSING_MESSAGE =
@@ -978,8 +987,8 @@ public class RCFile {
public Writer(FileSystem fs, Configuration conf, Path name, int bufferSize,
short replication, long blockSize, Progressable progress,
Metadata metadata, CompressionCodec codec) throws IOException {
- RECORD_INTERVAL = HiveConf.getIntVar(conf, HiveConf.ConfVars.HIVE_RCFILE_RECORD_INTERVAL);
- columnNumber = HiveConf.getIntVar(conf, HiveConf.ConfVars.HIVE_RCFILE_COLUMN_NUMBER_CONF);
+ RECORD_INTERVAL = HiveConf.getIntVar(conf, HIVE_RCFILE_RECORD_INTERVAL);
+ columnNumber = HiveConf.getIntVar(conf, HIVE_RCFILE_COLUMN_NUMBER_CONF);
if (metadata == null) {
metadata = new Metadata();
@@ -1051,8 +1060,7 @@ public class RCFile {
this.out = out;
this.codec = codec;
this.metadata = metadata;
- this.useNewMagic =
- conf.getBoolean(HiveConf.ConfVars.HIVEUSEEXPLICITRCFILEHEADER.varname, true);
+ this.useNewMagic = conf.getBoolean(HIVEUSEEXPLICITRCFILEHEADER.varname, true);
}
/** Returns the compression codec of data in this file. */
@@ -1339,8 +1347,7 @@ public class RCFile {
/** Create a new RCFile reader. */
public Reader(FileSystem fs, Path file, int bufferSize, Configuration conf,
long start, long length) throws IOException {
- tolerateCorruptions = HiveConf.getBoolVar(
- conf, HiveConf.ConfVars.HIVE_RCFILE_TOLERATE_CORRUPTIONS);
+ tolerateCorruptions = HiveConf.getBoolVar(conf, HIVE_RCFILE_TOLERATE_CORRUPTIONS);
conf.setInt("io.file.buffer.size", bufferSize);
this.file = file;
in = openFile(fs, file, bufferSize, length);
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ReduceSinkMapJoinProc.java Tue Jul 1 02:15:23 2014
@@ -117,6 +117,10 @@ public class ReduceSinkMapJoinProc imple
int numBuckets = -1;
EdgeType edgeType = EdgeType.BROADCAST_EDGE;
if (mapJoinOp.getConf().isBucketMapJoin()) {
+
+ // disable auto parallelism for bucket map joins
+ parentRS.getConf().setAutoParallel(false);
+
numBuckets = (Integer) mapJoinOp.getConf().getBigTableBucketNumMapping().values().toArray()[0];
if (mapJoinOp.getConf().getCustomBucketMapJoin()) {
edgeType = EdgeType.CUSTOM_EDGE;
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java Tue Jul 1 02:15:23 2014
@@ -19,8 +19,7 @@
package org.apache.hadoop.hive.ql.parse;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.LinkedHashMap;
+import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -33,7 +32,6 @@ import org.apache.hadoop.hive.ql.Context
import org.apache.hadoop.hive.ql.ErrorMsg;
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.metadata.HiveException;
-import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
/**
@@ -56,50 +54,8 @@ public class ColumnStatsSemanticAnalyzer
private boolean isTableLevel;
private List<String> colNames;
private List<String> colType;
- private String partName;
private Table tbl;
- private class PartitionList {
- private final String[] partKeys;
- private final String[] partValues;
- private final int numPartitions;
- private int numPartitionValues;
-
- PartitionList(int numPartitions) {
- this.numPartitions = numPartitions;
- partKeys = new String[numPartitions];
- partValues = new String[numPartitions];
- }
-
- public int getNumPartitions() {
- return numPartitions;
- }
-
- public String[] getPartValues() {
- return partValues;
- }
-
- public String[] getPartKeys() {
- return partKeys;
- }
-
- public void addPartValue(String partValue, int index) {
- partValues[index] = new String(partValue);
- }
-
- public void addPartKey(String partKey, int index) {
- partKeys[index] = new String(partKey);
- }
-
- public int getNumPartValues() {
- return numPartitionValues;
- }
-
- public void setNumPartValues(int numPartValues) {
- numPartitionValues = numPartValues;
- }
- }
-
public ColumnStatsSemanticAnalyzer(HiveConf conf) throws SemanticException {
super(conf, false);
}
@@ -143,29 +99,22 @@ public class ColumnStatsSemanticAnalyzer
}
}
- private PartitionList getPartKeyValuePairsFromAST(ASTNode tree) {
+ private Map<String,String> getPartKeyValuePairsFromAST(ASTNode tree) {
ASTNode child = ((ASTNode) tree.getChild(0).getChild(1));
- int numParts = child.getChildCount();
- PartitionList partList = new PartitionList(numParts);
+ Map<String,String> partSpec = new HashMap<String, String>();
String partKey;
String partValue;
- int numPartValue = 0;
- for (int i = 0; i < numParts; i++) {
- partKey = new String(getUnescapedName((ASTNode) child.getChild(i).getChild(0)));
+ for (int i = 0; i < child.getChildCount(); i++) {
+ partKey = new String(getUnescapedName((ASTNode) child.getChild(i).getChild(0))).toLowerCase();
if (child.getChild(i).getChildCount() > 1) {
partValue = new String(getUnescapedName((ASTNode) child.getChild(i).getChild(1)));
partValue = partValue.replaceAll("'", "");
- numPartValue += 1;
} else {
partValue = null;
}
- partList.addPartKey(partKey, i);
- if (partValue != null) {
- partList.addPartValue(partValue, i);
- }
+ partSpec.put(partKey, partValue);
}
- partList.setNumPartValues(numPartValue);
- return partList;
+ return partSpec;
}
private List<String> getColumnName(ASTNode tree) throws SemanticException{
@@ -186,133 +135,85 @@ public class ColumnStatsSemanticAnalyzer
}
}
- private void validatePartitionKeys(PartitionList partList) throws
+ private void handlePartialPartitionSpec(Map<String,String> partSpec) throws
SemanticException {
- List<FieldSchema> partKeys = tbl.getPartitionKeys();
- String[] inputPartKeys = partList.getPartKeys();
-
- if (inputPartKeys.length != partKeys.size()) {
- throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INCORRECT_NUM_PART_KEY.getMsg());
- }
-
- Map<String, Integer> partKeysMap = new LinkedHashMap<String, Integer>();
- for (int i=0; i <inputPartKeys.length; i++) {
- partKeysMap.put(inputPartKeys[i].toLowerCase(), new Integer(1));
- }
- // Verify that the user specified part keys match the part keys in the table
- for (FieldSchema partKey:partKeys) {
- if (!partKeysMap.containsKey(partKey.getName().toLowerCase())) {
- throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INVALID_PART_KEY.getMsg());
- }
- }
- }
-
- private String[] getPartitionKeysType(PartitionList partList) throws
- SemanticException {
- List<FieldSchema> partKeys = tbl.getPartitionKeys();
- String[] inputPartKeys = partList.getPartKeys();
- String[] inputPartKeyTypes = new String[inputPartKeys.length];
-
- for (int i=0; i < inputPartKeys.length; i++) {
- for (FieldSchema partKey:partKeys) {
- if (inputPartKeys[i].equalsIgnoreCase(partKey.getName())) {
- inputPartKeyTypes[i] = new String(partKey.getType());
- break;
- }
- }
- }
- return inputPartKeyTypes;
- }
-
- private String constructPartitionName(PartitionList partList)
- throws SemanticException {
- Partition part;
- String[] partKeys = partList.getPartKeys();
- String[] partValues = partList.getPartValues();
- Map<String, String> partSpec = new LinkedHashMap<String, String>();
-
- for (int i=0; i<partKeys.length; i++) {
- partSpec.put(partKeys[i].toLowerCase(), partValues[i]);
+ // If user has fully specified partition, validate that partition exists
+ int partValsSpecified = 0;
+ for (String partKey : partSpec.keySet()) {
+ partValsSpecified += partSpec.get(partKey) == null ? 0 : 1;
}
-
try {
- part = db.getPartition(tbl, partSpec, false);
- } catch (HiveException e) {
- throw new SemanticException(ErrorMsg.INVALID_PARTITION.getMsg(partName));
+ if ((partValsSpecified == tbl.getPartitionKeys().size()) && (db.getPartition(tbl, partSpec, false, null, false) == null)) {
+ throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INVALID_PARTITION.getMsg() + " : " + partSpec);
+ }
+ } catch (HiveException he) {
+ throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INVALID_PARTITION.getMsg() + " : " + partSpec);
}
- if (part == null) {
- throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INVALID_PARTITION.getMsg());
- }
- return part.getName();
+ // User might have only specified partial list of partition keys, in which case add other partition keys in partSpec
+ List<String> partKeys = Utilities.getColumnNamesFromFieldSchema(tbl.getPartitionKeys());
+ for (String partKey : partKeys){
+ if(!partSpec.containsKey(partKey)) {
+ partSpec.put(partKey, null);
+ }
+ }
+
+ // Check if user have erroneously specified non-existent partitioning columns
+ for (String partKey : partSpec.keySet()) {
+ if(!partKeys.contains(partKey)){
+ throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INVALID_PART_KEY.getMsg() + " : " + partKey);
+ }
+ }
}
- private void validatePartitionClause(PartitionList partList) throws
- SemanticException {
- int numPartKeys = partList.getNumPartitions();
- int numPartValues = partList.getNumPartValues();
- // Raise error if the user has specified dynamic partitions in the partitioning clause
- if (numPartKeys != numPartValues) {
- throw new SemanticException(ErrorMsg.COLUMNSTATSCOLLECTOR_INVALID_SYNTAX.getMsg());
- }
- // Validate the user specified partition keys match the partition keys in the table
- validatePartitionKeys(partList);
- }
-
- private StringBuilder genPartitionClause(PartitionList partList) throws SemanticException {
+ private StringBuilder genPartitionClause(Map<String,String> partSpec) throws SemanticException {
StringBuilder whereClause = new StringBuilder(" where ");
boolean predPresent = false;
StringBuilder groupByClause = new StringBuilder(" group by ");
boolean aggPresent = false;
- StringBuilder retClause = null;
- String[] partKeys = partList.getPartKeys();
- String[] partValues = partList.getPartValues();
- String[] partKeysType = getPartitionKeysType(partList);
- for (int i = 0; i < partList.getNumPartitions(); i++) {
- if (partValues[i] != null) {
+ for (String partKey : partSpec.keySet()) {
+ String value;
+ if ((value = partSpec.get(partKey)) != null) {
if (!predPresent) {
- whereClause.append(partKeys[i]);
- whereClause.append(" = ");
- if (partKeysType[i].equalsIgnoreCase("string")) {
- whereClause.append("'");
- }
- whereClause.append(partValues[i]);
- if (partKeysType[i].equalsIgnoreCase("string")) {
- whereClause.append("'");
- }
predPresent = true;
} else {
whereClause.append(" and ");
- whereClause.append(partKeys[i]);
- whereClause.append(" = ");
- if (partKeysType[i].equalsIgnoreCase("string")) {
- whereClause.append("'");
- }
- whereClause.append(partValues[i]);
- if (partKeysType[i].equalsIgnoreCase("string")) {
- whereClause.append("'");
- }
}
- } else {
+ whereClause.append(partKey);
+ whereClause.append(" = ");
+ if (getColTypeOf(partKey).equalsIgnoreCase("string")) {
+ whereClause.append("'");
+ }
+ whereClause.append(value);
+ if (getColTypeOf(partKey).equalsIgnoreCase("string")) {
+ whereClause.append("'");
+ }
+ }
+ }
+
+ for (FieldSchema fs : tbl.getPartitionKeys()) {
if (!aggPresent) {
- groupByClause.append(partKeys[i]);
aggPresent = true;
} else {
groupByClause.append(",");
- groupByClause.append(partKeys[i]);
}
- }
+ groupByClause.append(fs.getName());
}
+
// attach the predicate and group by to the return clause
- if (predPresent) {
- retClause = new StringBuilder(whereClause);
- }
- if (aggPresent) {
- retClause.append(groupByClause);
+ return predPresent ? whereClause.append(groupByClause) : groupByClause;
+ }
+
+ private String getColTypeOf (String partKey) throws SemanticException{
+
+ for (FieldSchema fs : tbl.getPartitionKeys()) {
+ if (partKey.equalsIgnoreCase(fs.getName())) {
+ return fs.getType();
+ }
}
- return retClause;
+ throw new SemanticException ("Unknown partition key : " + partKey);
}
private int getNumBitVectorsForNDVEstimation(HiveConf conf) throws SemanticException {
@@ -369,7 +270,7 @@ public class ColumnStatsSemanticAnalyzer
return numBitVectors;
}
- private List<String> getTableColumnType(List<String> colNames)
+ private List<String> getColumnTypes(List<String> colNames)
throws SemanticException{
List<String> colTypes = new LinkedList<String>();
List<FieldSchema> cols = tbl.getCols();
@@ -384,35 +285,11 @@ public class ColumnStatsSemanticAnalyzer
return colTypes;
}
- private List<String> getPartitionColumnType(String partName,
- List<String> colNames) throws SemanticException {
- List<String> colTypes = new LinkedList<String>();
- List<String> partNames = new ArrayList<String>();
- partNames.add(partName);
- List<Partition> partitionList;
-
- try {
- partitionList = db.getPartitionsByNames(tbl, partNames);
- } catch (HiveException e) {
- throw new SemanticException(ErrorMsg.INVALID_PARTITION.getMsg(partName));
- }
- Partition part = partitionList.get(0);
- List<FieldSchema> cols = part.getCols();
-
- for (String colName : colNames) {
- for (FieldSchema col: cols) {
- if (colName.equalsIgnoreCase(col.getName())) {
- colTypes.add(new String(col.getType()));
- }
- }
- }
- return colTypes;
- }
-
- private String genRewrittenQuery(List<String> colNames, int numBitVectors, PartitionList partList,
+ private String genRewrittenQuery(List<String> colNames, int numBitVectors, Map<String,String> partSpec,
boolean isPartitionStats) throws SemanticException{
StringBuilder rewrittenQueryBuilder = new StringBuilder("select ");
String rewrittenQuery;
+
for (int i = 0; i < colNames.size(); i++) {
if (i > 0) {
rewrittenQueryBuilder.append(" , ");
@@ -423,6 +300,9 @@ public class ColumnStatsSemanticAnalyzer
rewrittenQueryBuilder.append(numBitVectors);
rewrittenQueryBuilder.append(" )");
}
+ for (FieldSchema fs : tbl.getPartCols()) {
+ rewrittenQueryBuilder.append(" , " + fs.getName());
+ }
rewrittenQueryBuilder.append(" from ");
rewrittenQueryBuilder.append(tbl.getTableName());
isRewritten = true;
@@ -430,7 +310,7 @@ public class ColumnStatsSemanticAnalyzer
// If partition level statistics is requested, add predicate and group by as needed to rewritten
// query
if (isPartitionStats) {
- rewrittenQueryBuilder.append(genPartitionClause(partList));
+ rewrittenQueryBuilder.append(genPartitionClause(partSpec));
}
rewrittenQuery = rewrittenQueryBuilder.toString();
@@ -474,23 +354,20 @@ public class ColumnStatsSemanticAnalyzer
// Save away the original AST
originalTree = tree;
boolean isPartitionStats = isPartitionLevelStats(tree);
- PartitionList partList = null;
+ Map<String,String> partSpec = null;
checkForPartitionColumns(colNames, Utilities.getColumnNamesFromFieldSchema(tbl.getPartitionKeys()));
validateSpecifiedColumnNames(colNames);
if (isPartitionStats) {
isTableLevel = false;
- partList = getPartKeyValuePairsFromAST(tree);
- validatePartitionClause(partList);
- partName = constructPartitionName(partList);
- colType = getPartitionColumnType(partName, colNames);
+ partSpec = getPartKeyValuePairsFromAST(tree);
+ handlePartialPartitionSpec(partSpec);
} else {
isTableLevel = true;
- colType = getTableColumnType(colNames);
}
-
+ colType = getColumnTypes(colNames);
int numBitVectors = getNumBitVectorsForNDVEstimation(conf);
- rewrittenQuery = genRewrittenQuery(colNames, numBitVectors, partList, isPartitionStats);
+ rewrittenQuery = genRewrittenQuery(colNames, numBitVectors, partSpec, isPartitionStats);
rewrittenTree = genRewrittenTree(rewrittenQuery);
} else {
// Not an analyze table column compute statistics statement - don't do any rewrites
@@ -540,10 +417,6 @@ public class ColumnStatsSemanticAnalyzer
qbp = qb.getParseInfo();
qbp.setTableName(tbl.getTableName());
qbp.setTblLvl(isTableLevel);
-
- if (!isTableLevel) {
- qbp.setPartName(partName);
- }
qbp.setColName(colNames);
qbp.setColType(colType);
initCtx(ctx);
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/QBParseInfo.java Tue Jul 1 02:15:23 2014
@@ -65,7 +65,6 @@ public class QBParseInfo {
private String tableName; // used for column statistics
private List<String> colName; // used for column statistics
private List<String> colType; // used for column statistics
- private String partName; // used for column statistics
private boolean isTblLvl; // used for column statistics
@@ -582,14 +581,6 @@ public class QBParseInfo {
this.colName = colName;
}
- public String getPartName() {
- return partName;
- }
-
- public void setPartName(String partName) {
- this.partName = partName;
- }
-
public boolean isTblLvl() {
return isTblLvl;
}
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=1606964&r1=1606963&r2=1606964&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 Tue Jul 1 02:15:23 2014
@@ -67,6 +67,7 @@ import org.apache.hadoop.hive.ql.QueryPr
import org.apache.hadoop.hive.ql.exec.AbstractMapJoinOperator;
import org.apache.hadoop.hive.ql.exec.ArchiveUtils;
import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.exec.ExprNodeEvaluatorFactory;
import org.apache.hadoop.hive.ql.exec.FetchTask;
import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
import org.apache.hadoop.hive.ql.exec.FunctionInfo;
@@ -4224,6 +4225,8 @@ public class SemanticAnalyzer extends Ba
reduceSinkInputRowResolver, reduceSinkOutputRowResolver, outputKeyColumnNames,
colExprMap);
+ int keyLength = reduceKeys.size();
+
// add a key for reduce sink
if (groupingSetsPresent) {
// Process grouping set for the reduce sink operator
@@ -4275,7 +4278,7 @@ public class SemanticAnalyzer extends Ba
ReduceSinkOperator rsOp = (ReduceSinkOperator) putOpInsertMap(
OperatorFactory.getAndMakeChild(
PlanUtils.getReduceSinkDesc(reduceKeys,
- groupingSetsPresent ? grpByExprs.size() + 1 : grpByExprs.size(),
+ groupingSetsPresent ? keyLength + 1 : keyLength,
reduceValues, distinctColIndices,
outputKeyColumnNames, outputValueColumnNames, true, -1, numPartitionFields,
numReducers),
@@ -4296,24 +4299,32 @@ public class SemanticAnalyzer extends Ba
ASTNode grpbyExpr = grpByExprs.get(i);
ExprNodeDesc inputExpr = genExprNodeDesc(grpbyExpr,
reduceSinkInputRowResolver);
- reduceKeys.add(inputExpr);
- if (reduceSinkOutputRowResolver.getExpression(grpbyExpr) == null) {
- outputKeyColumnNames.add(getColumnInternalName(reduceKeys.size() - 1));
- String field = Utilities.ReduceField.KEY.toString() + "."
- + getColumnInternalName(reduceKeys.size() - 1);
- ColumnInfo colInfo = new ColumnInfo(field, reduceKeys.get(
- reduceKeys.size() - 1).getTypeInfo(), null, false);
- reduceSinkOutputRowResolver.putExpression(grpbyExpr, colInfo);
- colExprMap.put(colInfo.getInternalName(), inputExpr);
- } else {
- throw new SemanticException(ErrorMsg.DUPLICATE_GROUPBY_KEY
- .getMsg(grpbyExpr));
+ ColumnInfo prev = reduceSinkOutputRowResolver.getExpression(grpbyExpr);
+ if (prev != null && isDeterministic(inputExpr)) {
+ colExprMap.put(prev.getInternalName(), inputExpr);
+ continue;
}
+ reduceKeys.add(inputExpr);
+ outputKeyColumnNames.add(getColumnInternalName(reduceKeys.size() - 1));
+ String field = Utilities.ReduceField.KEY.toString() + "."
+ + getColumnInternalName(reduceKeys.size() - 1);
+ ColumnInfo colInfo = new ColumnInfo(field, reduceKeys.get(
+ reduceKeys.size() - 1).getTypeInfo(), null, false);
+ reduceSinkOutputRowResolver.putExpression(grpbyExpr, colInfo);
+ colExprMap.put(colInfo.getInternalName(), inputExpr);
}
return reduceKeys;
}
+ private boolean isDeterministic(ExprNodeDesc expr) throws SemanticException {
+ try {
+ return ExprNodeEvaluatorFactory.get(expr).isDeterministic();
+ } catch (Exception e) {
+ throw new SemanticException(e);
+ }
+ }
+
private List<List<Integer>> getDistinctColIndicesForReduceSink(QBParseInfo parseInfo,
String dest,
List<ExprNodeDesc> reduceKeys, RowResolver reduceSinkInputRowResolver,
@@ -4419,6 +4430,8 @@ public class SemanticAnalyzer extends Ba
reduceSinkInputRowResolver, reduceSinkOutputRowResolver, outputKeyColumnNames,
colExprMap);
+ int keyLength = reduceKeys.size();
+
List<List<Integer>> distinctColIndices = getDistinctColIndicesForReduceSink(parseInfo, dest,
reduceKeys, reduceSinkInputRowResolver, reduceSinkOutputRowResolver, outputKeyColumnNames,
colExprMap);
@@ -4466,8 +4479,8 @@ public class SemanticAnalyzer extends Ba
ReduceSinkOperator rsOp = (ReduceSinkOperator) putOpInsertMap(
OperatorFactory.getAndMakeChild(PlanUtils.getReduceSinkDesc(reduceKeys,
- grpByExprs.size(), reduceValues, distinctColIndices,
- outputKeyColumnNames, outputValueColumnNames, true, -1, grpByExprs.size(),
+ keyLength, reduceValues, distinctColIndices,
+ outputKeyColumnNames, outputValueColumnNames, true, -1, keyLength,
-1), new RowSchema(reduceSinkOutputRowResolver
.getColumnInfos()), inputOperatorInfo), reduceSinkOutputRowResolver);
rsOp.setColumnExprMap(colExprMap);
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/TaskCompiler.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/TaskCompiler.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/TaskCompiler.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/parse/TaskCompiler.java Tue Jul 1 02:15:23 2014
@@ -293,7 +293,6 @@ public abstract class TaskCompiler {
ColumnStatsWork cStatsWork = null;
FetchWork fetch = null;
String tableName = qbParseInfo.getTableName();
- String partName = qbParseInfo.getPartName();
List<String> colName = qbParseInfo.getColName();
List<String> colType = qbParseInfo.getColType();
boolean isTblLevel = qbParseInfo.isTblLvl();
@@ -307,7 +306,7 @@ public abstract class TaskCompiler {
fetch = new FetchWork(loadFileWork.get(0).getSourcePath(),
resultTab, qb.getParseInfo().getOuterQueryLimit());
- ColumnStatsDesc cStatsDesc = new ColumnStatsDesc(tableName, partName,
+ ColumnStatsDesc cStatsDesc = new ColumnStatsDesc(tableName,
colName, colType, isTblLevel);
cStatsWork = new ColumnStatsWork(fetch, cStatsDesc);
cStatsTask = (ColumnStatsTask) TaskFactory.get(cStatsWork, conf);
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ColumnStatsDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ColumnStatsDesc.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ColumnStatsDesc.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ColumnStatsDesc.java Tue Jul 1 02:15:23 2014
@@ -28,19 +28,17 @@ public class ColumnStatsDesc extends DDL
private static final long serialVersionUID = 1L;
private boolean isTblLevel;
private String tableName;
- private String partName;
private List<String> colName;
private List<String> colType;
public ColumnStatsDesc() {
}
- public ColumnStatsDesc(String tableName, String partName, List<String> colName,
+ public ColumnStatsDesc(String tableName, List<String> colName,
List<String> colType, boolean isTblLevel) {
this.tableName = tableName;
this.colName = colName;
this.colType = colType;
- this.partName = partName;
this.isTblLevel = isTblLevel;
}
@@ -62,15 +60,6 @@ public class ColumnStatsDesc extends DDL
this.isTblLevel = isTblLevel;
}
- @Explain(displayName = "Partition")
- public String getPartName() {
- return partName;
- }
-
- public void setPartName(String partName) {
- this.partName = partName;
- }
-
@Explain(displayName = "Columns")
public List<String> getColName() {
return colName;
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/PlanUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/PlanUtils.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/PlanUtils.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/PlanUtils.java Tue Jul 1 02:15:23 2014
@@ -739,20 +739,15 @@ public final class PlanUtils {
List<String> outputKeyColumnNames, List<String> outputValueColumnNames,
boolean includeKey, int tag,
int numPartitionFields, int numReducers) throws SemanticException {
- ArrayList<ExprNodeDesc> partitionCols = null;
+ ArrayList<ExprNodeDesc> partitionCols = new ArrayList<ExprNodeDesc>();
if (numPartitionFields >= keyCols.size()) {
- partitionCols = keyCols;
+ partitionCols.addAll(keyCols);
} else if (numPartitionFields >= 0) {
- partitionCols = new ArrayList<ExprNodeDesc>(numPartitionFields);
- for (int i = 0; i < numPartitionFields; i++) {
- partitionCols.add(keyCols.get(i));
- }
+ partitionCols.addAll(keyCols.subList(0, numPartitionFields));
} else {
// numPartitionFields = -1 means random partitioning
- partitionCols = new ArrayList<ExprNodeDesc>(1);
- partitionCols.add(TypeCheckProcFactory.DefaultExprProcessor
- .getFuncExprNodeDesc("rand"));
+ partitionCols.add(TypeCheckProcFactory.DefaultExprProcessor.getFuncExprNodeDesc("rand"));
}
StringBuilder order = new StringBuilder();
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ReduceSinkDesc.java Tue Jul 1 02:15:23 2014
@@ -86,8 +86,10 @@ public class ReduceSinkDesc extends Abst
private int topN = -1;
private float topNMemoryUsage = -1;
private boolean mapGroupBy; // for group-by, values with same key on top-K should be forwarded
+ //flag used to control how TopN handled for PTF/Windowing partitions.
+ private boolean isPTFReduceSink = false;
private boolean skipTag; // Skip writing tags when feeding into mapjoin hashtable
- private boolean autoParallel = false; // Is reducer parallelism automatic or fixed
+ private Boolean autoParallel = null; // Is reducer auto-parallelism enabled, disabled or unset
private static transient Log LOG = LogFactory.getLog(ReduceSinkDesc.class);
public ReduceSinkDesc() {
@@ -140,7 +142,7 @@ public class ReduceSinkDesc extends Abst
desc.setBucketCols(bucketCols);
desc.setStatistics(this.getStatistics());
desc.setSkipTag(skipTag);
- desc.setAutoParallel(autoParallel);
+ desc.autoParallel = autoParallel;
return desc;
}
@@ -253,6 +255,14 @@ public class ReduceSinkDesc extends Abst
this.mapGroupBy = mapGroupBy;
}
+ public boolean isPTFReduceSink() {
+ return isPTFReduceSink;
+ }
+
+ public void setPTFReduceSink(boolean isPTFReduceSink) {
+ this.isPTFReduceSink = isPTFReduceSink;
+ }
+
/**
* Returns the number of reducers for the map-reduce job. -1 means to decide
* the number of reducers at runtime. This enables Hive to estimate the number
@@ -344,10 +354,16 @@ public class ReduceSinkDesc extends Abst
}
public final boolean isAutoParallel() {
- return autoParallel;
+ return (autoParallel != null) && autoParallel;
}
public final void setAutoParallel(final boolean autoParallel) {
- this.autoParallel = autoParallel;
+ // we don't allow turning on auto parallel once it has been
+ // explicitly turned off. That is to avoid scenarios where
+ // auto parallelism could break assumptions about number of
+ // reducers or hash function.
+ if (this.autoParallel == null || this.autoParallel == true) {
+ this.autoParallel = autoParallel;
+ }
}
}
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ptf/WindowTableFunctionDef.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ptf/WindowTableFunctionDef.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ptf/WindowTableFunctionDef.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/plan/ptf/WindowTableFunctionDef.java Tue Jul 1 02:15:23 2014
@@ -23,6 +23,9 @@ import java.util.List;
public class WindowTableFunctionDef extends PartitionedTableFunctionDef {
List<WindowFunctionDef> windowFunctions;
+
+ int rankLimit = -1;
+ int rankLimitFunction;
public List<WindowFunctionDef> getWindowFunctions() {
return windowFunctions;
@@ -30,4 +33,16 @@ public class WindowTableFunctionDef exte
public void setWindowFunctions(List<WindowFunctionDef> windowFunctions) {
this.windowFunctions = windowFunctions;
}
+ public int getRankLimit() {
+ return rankLimit;
+ }
+ public void setRankLimit(int rankLimit) {
+ this.rankLimit = rankLimit;
+ }
+ public int getRankLimitFunction() {
+ return rankLimitFunction;
+ }
+ public void setRankLimitFunction(int rankLimitFunction) {
+ this.rankLimitFunction = rankLimitFunction;
+ }
}
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpProcFactory.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpProcFactory.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpProcFactory.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/ppd/OpProcFactory.java Tue Jul 1 02:15:23 2014
@@ -37,6 +37,7 @@ import org.apache.hadoop.hive.ql.exec.Jo
import org.apache.hadoop.hive.ql.exec.LateralViewJoinOperator;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.OperatorFactory;
+import org.apache.hadoop.hive.ql.exec.PTFOperator;
import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.exec.RowSchema;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
@@ -53,7 +54,9 @@ import org.apache.hadoop.hive.ql.parse.H
import org.apache.hadoop.hive.ql.parse.OpParseContext;
import org.apache.hadoop.hive.ql.parse.RowResolver;
import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.parse.WindowingSpec.Direction;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
+import org.apache.hadoop.hive.ql.plan.ExprNodeConstantDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDescUtils;
import org.apache.hadoop.hive.ql.plan.ExprNodeGenericFuncDesc;
@@ -61,8 +64,23 @@ import org.apache.hadoop.hive.ql.plan.Fi
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.PTFDesc;
+import org.apache.hadoop.hive.ql.plan.ReduceSinkDesc;
import org.apache.hadoop.hive.ql.plan.TableScanDesc;
+import org.apache.hadoop.hive.ql.plan.ptf.BoundaryDef;
+import org.apache.hadoop.hive.ql.plan.ptf.ValueBoundaryDef;
+import org.apache.hadoop.hive.ql.plan.ptf.WindowFrameDef;
+import org.apache.hadoop.hive.ql.plan.ptf.WindowFunctionDef;
+import org.apache.hadoop.hive.ql.plan.ptf.WindowTableFunctionDef;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFDenseRank.GenericUDAFDenseRankEvaluator;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFLead.GenericUDAFLeadEvaluator;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFRank.GenericUDAFRankEvaluator;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPEqualOrLessThan;
+import org.apache.hadoop.hive.ql.udf.generic.GenericUDFOPLessThan;
+import org.apache.hadoop.hive.ql.udf.ptf.WindowingTableFunction;
import org.apache.hadoop.hive.serde2.Deserializer;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.mapred.JobConf;
/**
@@ -133,6 +151,191 @@ public final class OpProcFactory {
}
}
+
+ public static class PTFPPD extends ScriptPPD {
+
+ /*
+ * For WindowingTableFunction if:
+ * a. there is a Rank/DenseRank function: if there are unpushedPred of the form
+ * rnkValue < Constant; then use the smallest Constant val as the 'rankLimit'
+ * on the WindowingTablFn.
+ * b. If there are no Wdw Fns with an End Boundary past the current row, the
+ * condition can be pushed down as a limit pushdown(mapGroupBy=true)
+ *
+ * (non-Javadoc)
+ * @see org.apache.hadoop.hive.ql.ppd.OpProcFactory.ScriptPPD#process(org.apache.hadoop.hive.ql.lib.Node, java.util.Stack, org.apache.hadoop.hive.ql.lib.NodeProcessorCtx, java.lang.Object[])
+ */
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ LOG.info("Processing for " + nd.getName() + "("
+ + ((Operator) nd).getIdentifier() + ")");
+ OpWalkerInfo owi = (OpWalkerInfo) procCtx;
+ PTFOperator ptfOp = (PTFOperator) nd;
+
+ pushRankLimit(ptfOp, owi);
+ return super.process(nd, stack, procCtx, nodeOutputs);
+ }
+
+ private void pushRankLimit(PTFOperator ptfOp, OpWalkerInfo owi) throws SemanticException {
+ PTFDesc conf = ptfOp.getConf();
+
+ if ( !conf.forWindowing() ) {
+ return;
+ }
+
+ float threshold = owi.getParseContext().getConf().getFloatVar(HiveConf.ConfVars.HIVELIMITPUSHDOWNMEMORYUSAGE);
+ if (threshold <= 0 || threshold >= 1) {
+ return;
+ }
+
+ WindowTableFunctionDef wTFn = (WindowTableFunctionDef) conf.getFuncDef();
+ List<Integer> rFnIdxs = rankingFunctions(wTFn);
+
+ if ( rFnIdxs.size() == 0 ) {
+ return;
+ }
+
+ ExprWalkerInfo childInfo = getChildWalkerInfo((Operator<?>) ptfOp, owi);
+
+ if (childInfo == null) {
+ return;
+ }
+
+ List<ExprNodeDesc> preds = new ArrayList<ExprNodeDesc>();
+ Iterator<List<ExprNodeDesc>> iterator = childInfo.getFinalCandidates().values().iterator();
+ while (iterator.hasNext()) {
+ for (ExprNodeDesc pred : iterator.next()) {
+ preds = ExprNodeDescUtils.split(pred, preds);
+ }
+ }
+
+ int rLimit = -1;
+ int fnIdx = -1;
+ for(ExprNodeDesc pred : preds) {
+ int[] pLimit = getLimit(wTFn, rFnIdxs, pred);
+ if ( pLimit != null ) {
+ if ( rLimit == -1 || rLimit >= pLimit[0] ) {
+ rLimit = pLimit[0];
+ fnIdx = pLimit[1];
+ }
+ }
+ }
+
+ if ( rLimit != -1 ) {
+ wTFn.setRankLimit(rLimit);
+ wTFn.setRankLimitFunction(fnIdx);
+ if ( canPushLimitToReduceSink(wTFn)) {
+ pushRankLimitToRedSink(ptfOp, owi.getParseContext().getConf(), rLimit);
+ }
+ }
+ }
+
+ private List<Integer> rankingFunctions(WindowTableFunctionDef wTFn) {
+ List<Integer> rFns = new ArrayList<Integer>();
+ for(int i=0; i < wTFn.getWindowFunctions().size(); i++ ) {
+ WindowFunctionDef wFnDef = wTFn.getWindowFunctions().get(i);
+ if ( (wFnDef.getWFnEval() instanceof GenericUDAFRankEvaluator) ||
+ (wFnDef.getWFnEval() instanceof GenericUDAFDenseRankEvaluator ) ) {
+ rFns.add(i);
+ }
+ }
+ return rFns;
+ }
+
+ /*
+ * For a predicate check if it is a candidate for pushing down as limit optimization.
+ * The expression must be of the form rankFn <|<= constant.
+ */
+ private int[] getLimit(WindowTableFunctionDef wTFn, List<Integer> rFnIdxs, ExprNodeDesc expr) {
+
+ if ( !(expr instanceof ExprNodeGenericFuncDesc) ) {
+ return null;
+ }
+
+ ExprNodeGenericFuncDesc fExpr = (ExprNodeGenericFuncDesc) expr;
+
+ if ( !(fExpr.getGenericUDF() instanceof GenericUDFOPLessThan) &&
+ !(fExpr.getGenericUDF() instanceof GenericUDFOPEqualOrLessThan) ) {
+ return null;
+ }
+
+ if ( !(fExpr.getChildren().get(0) instanceof ExprNodeColumnDesc) ) {
+ return null;
+ }
+
+ if ( !(fExpr.getChildren().get(1) instanceof ExprNodeConstantDesc) ) {
+ return null;
+ }
+
+ ExprNodeConstantDesc constantExpr = (ExprNodeConstantDesc) fExpr.getChildren().get(1) ;
+
+ if ( constantExpr.getTypeInfo() != TypeInfoFactory.intTypeInfo ) {
+ return null;
+ }
+
+ int limit = (Integer) constantExpr.getValue();
+ if ( fExpr.getGenericUDF() instanceof GenericUDFOPEqualOrLessThan ) {
+ limit = limit + 1;
+ }
+ String colName = ((ExprNodeColumnDesc)fExpr.getChildren().get(0)).getColumn();
+
+ for(int i=0; i < rFnIdxs.size(); i++ ) {
+ String fAlias = wTFn.getWindowFunctions().get(i).getAlias();
+ if ( fAlias.equals(colName)) {
+ return new int[] {limit,i};
+ }
+ }
+
+ return null;
+ }
+
+ /*
+ * Limit can be pushed down to Map-side if all Window Functions need access
+ * to rows before the current row. This is true for:
+ * 1. Rank, DenseRank and Lead Fns. (the window doesn't matter for lead fn).
+ * 2. If the Window for the function is Row based and the End Boundary doesn't
+ * reference rows past the Current Row.
+ */
+ private boolean canPushLimitToReduceSink(WindowTableFunctionDef wTFn) {
+
+ for(WindowFunctionDef wFnDef : wTFn.getWindowFunctions() ) {
+ if ( (wFnDef.getWFnEval() instanceof GenericUDAFRankEvaluator) ||
+ (wFnDef.getWFnEval() instanceof GenericUDAFDenseRankEvaluator ) ||
+ (wFnDef.getWFnEval() instanceof GenericUDAFLeadEvaluator ) ) {
+ continue;
+ }
+ WindowFrameDef wdwFrame = wFnDef.getWindowFrame();
+ BoundaryDef end = wdwFrame.getEnd();
+ if ( end instanceof ValueBoundaryDef ) {
+ return false;
+ }
+ if ( end.getDirection() == Direction.FOLLOWING ) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void pushRankLimitToRedSink(PTFOperator ptfOp, HiveConf conf, int rLimit) throws SemanticException {
+
+ Operator<? extends OperatorDesc> parent = ptfOp.getParentOperators().get(0);
+ Operator<? extends OperatorDesc> gP = parent == null ? null : parent.getParentOperators().get(0);
+
+ if ( gP == null || !(gP instanceof ReduceSinkOperator )) {
+ return;
+ }
+
+ float threshold = conf.getFloatVar(HiveConf.ConfVars.HIVELIMITPUSHDOWNMEMORYUSAGE);
+
+ ReduceSinkOperator rSink = (ReduceSinkOperator) gP;
+ ReduceSinkDesc rDesc = rSink.getConf();
+ rDesc.setTopN(rLimit);
+ rDesc.setTopNMemoryUsage(threshold);
+ rDesc.setMapGroupBy(true);
+ rDesc.setPTFReduceSink(true);
+ }
+ }
public static class UDTFPPD extends DefaultPPD implements NodeProcessor {
@Override
@@ -865,7 +1068,7 @@ public final class OpProcFactory {
}
public static NodeProcessor getPTFProc() {
- return new ScriptPPD();
+ return new PTFPPD();
}
public static NodeProcessor getSCRProc() {
Modified: hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/udf/ptf/WindowingTableFunction.java
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/udf/ptf/WindowingTableFunction.java?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/udf/ptf/WindowingTableFunction.java (original)
+++ hive/branches/cbo/ql/src/java/org/apache/hadoop/hive/ql/udf/ptf/WindowingTableFunction.java Tue Jul 1 02:15:23 2014
@@ -62,6 +62,7 @@ import org.apache.hadoop.hive.serde2.obj
public class WindowingTableFunction extends TableFunctionEvaluator {
StreamingState streamingState;
+ RankLimit rnkLimitDef;
@SuppressWarnings({ "unchecked", "rawtypes" })
@Override
@@ -283,6 +284,12 @@ public class WindowingTableFunction exte
}
}
}
+
+ if ( tabDef.getRankLimit() != -1 ) {
+ rnkLimitDef = new RankLimit(tabDef.getRankLimit(),
+ tabDef.getRankLimitFunction(), tabDef.getWindowFunctions());
+ }
+
streamingState = new StreamingState(cfg, inputOI, isMapSide, tabDef,
span[0], span[1]);
}
@@ -313,6 +320,13 @@ public class WindowingTableFunction exte
@Override
public List<Object> processRow(Object row) throws HiveException {
+ /*
+ * Once enough rows have been output, there is no need to process input rows.
+ */
+ if ( streamingState.rankLimitReached() ) {
+ return null;
+ }
+
streamingState.rollingPart.append(row);
row = streamingState.rollingPart
.getAt(streamingState.rollingPart.size() - 1);
@@ -382,6 +396,13 @@ public class WindowingTableFunction exte
@Override
public List<Object> finishPartition() throws HiveException {
+ /*
+ * Once enough rows have been output, there is no need to generate more output.
+ */
+ if ( streamingState.rankLimitReached() ) {
+ return null;
+ }
+
WindowTableFunctionDef tabDef = (WindowTableFunctionDef) getTableDef();
for (int i = 0; i < tabDef.getWindowFunctions().size(); i++) {
WindowFunctionDef wFn = tabDef.getWindowFunctions().get(i);
@@ -428,15 +449,17 @@ public class WindowingTableFunction exte
List<Object> oRows = new ArrayList<Object>();
- while (!streamingState.rollingPart.processedAllRows()) {
+ while (!streamingState.rollingPart.processedAllRows() &&
+ !streamingState.rankLimitReached() ) {
boolean hasRow = streamingState.hasOutputRow();
- ;
- if (!hasRow) {
+ if (!hasRow && !streamingState.rankLimitReached() ) {
throw new HiveException(
"Internal Error: cannot generate all output rows for a Partition");
}
- oRows.add(streamingState.nextOutputRow());
+ if ( hasRow ) {
+ oRows.add(streamingState.nextOutputRow());
+ }
}
return oRows.size() == 0 ? null : oRows;
@@ -496,6 +519,11 @@ public class WindowingTableFunction exte
output.add(null);
}
+ if ( wTFnDef.getRankLimit() != -1 ) {
+ rnkLimitDef = new RankLimit(wTFnDef.getRankLimit(),
+ wTFnDef.getRankLimitFunction(), wTFnDef.getWindowFunctions());
+ }
+
return new WindowingIterator(iPart, output, outputFromPivotFunctions,
ArrayUtils.toPrimitive(wFnsWithWindows.toArray(new Integer[wFnsWithWindows.size()])));
}
@@ -1205,6 +1233,7 @@ public class WindowingTableFunction exte
StructObjectInspector inputOI;
AggregationBuffer[] aggBuffers;
Object[][] args;
+ RankLimit rnkLimit;
WindowingIterator(PTFPartition iPart, ArrayList<Object> output,
List<?>[] outputFromPivotFunctions, int[] wFnsToProcess) {
@@ -1229,10 +1258,17 @@ public class WindowingTableFunction exte
} catch (HiveException he) {
throw new RuntimeException(he);
}
+ if ( WindowingTableFunction.this.rnkLimitDef != null ) {
+ rnkLimit = new RankLimit(WindowingTableFunction.this.rnkLimitDef);
+ }
}
@Override
public boolean hasNext() {
+
+ if ( rnkLimit != null && rnkLimit.limitReached() ) {
+ return false;
+ }
return currIdx < iPart.size();
}
@@ -1279,6 +1315,9 @@ public class WindowingTableFunction exte
throw new RuntimeException(he);
}
+ if ( rnkLimit != null ) {
+ rnkLimit.updateRank(output);
+ }
currIdx++;
return output;
}
@@ -1296,6 +1335,7 @@ public class WindowingTableFunction exte
AggregationBuffer[] aggBuffers;
Object[][] funcArgs;
Order order;
+ RankLimit rnkLimit;
@SuppressWarnings("unchecked")
StreamingState(Configuration cfg, StructObjectInspector inputOI,
@@ -1321,6 +1361,9 @@ public class WindowingTableFunction exte
funcArgs[i] = new Object[wFn.getArgs() == null ? 0 : wFn.getArgs().size()];
aggBuffers[i] = wFn.getWFnEval().getNewAggregationBuffer();
}
+ if ( WindowingTableFunction.this.rnkLimitDef != null ) {
+ rnkLimit = new RankLimit(WindowingTableFunction.this.rnkLimitDef);
+ }
}
void reset(WindowTableFunctionDef tabDef) throws HiveException {
@@ -1334,9 +1377,17 @@ public class WindowingTableFunction exte
WindowFunctionDef wFn = tabDef.getWindowFunctions().get(i);
aggBuffers[i] = wFn.getWFnEval().getNewAggregationBuffer();
}
+
+ if ( rnkLimit != null ) {
+ rnkLimit.reset();
+ }
}
boolean hasOutputRow() {
+ if ( rankLimitReached() ) {
+ return false;
+ }
+
for (int i = 0; i < fnOutputs.length; i++) {
if (fnOutputs[i].size() == 0) {
return false;
@@ -1355,8 +1406,65 @@ public class WindowingTableFunction exte
for (StructField f : rollingPart.getOutputOI().getAllStructFieldRefs()) {
oRow.add(rollingPart.getOutputOI().getStructFieldData(iRow, f));
}
+ if ( rnkLimit != null ) {
+ rnkLimit.updateRank(oRow);
+ }
return oRow;
}
+
+ boolean rankLimitReached() {
+ return rnkLimit != null && rnkLimit.limitReached();
+ }
+ }
+
+ static class RankLimit {
+
+ /*
+ * Rows with a rank <= rankLimit are output.
+ * Only the first row with rank = rankLimit is output.
+ */
+ final int rankLimit;
+
+ /*
+ * the rankValue of the last row output.
+ */
+ int currentRank;
+
+ /*
+ * index of Rank function.
+ */
+ final int rankFnIdx;
+
+ final PrimitiveObjectInspector fnOutOI;
+
+ RankLimit(int rankLimit, int rankFnIdx, List<WindowFunctionDef> wdwFnDefs) {
+ this.rankLimit = rankLimit;
+ this.rankFnIdx = rankFnIdx;
+ this.fnOutOI = (PrimitiveObjectInspector) wdwFnDefs.get(rankFnIdx).getOI();
+ this.currentRank = -1;
+ }
+
+ RankLimit(RankLimit rl) {
+ this.rankLimit = rl.rankLimit;
+ this.rankFnIdx = rl.rankFnIdx;
+ this.fnOutOI = rl.fnOutOI;
+ this.currentRank = -1;
+ }
+
+ void reset() {
+ this.currentRank = -1;
+ }
+
+ void updateRank(List<Object> oRow) {
+ int r = (Integer) fnOutOI.getPrimitiveJavaObject(oRow.get(rankFnIdx));
+ if ( r > currentRank ) {
+ currentRank = r;
+ }
+ }
+
+ boolean limitReached() {
+ return currentRank >= rankLimit;
+ }
}
}
Modified: hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_invalid_values.q.out
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_invalid_values.q.out?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_invalid_values.q.out (original)
+++ hive/branches/cbo/ql/src/test/results/clientnegative/columnstats_partlvl_invalid_values.q.out Tue Jul 1 02:15:23 2014
@@ -65,4 +65,4 @@ POSTHOOK: type: LOAD
#### A masked pattern was here ####
POSTHOOK: Output: default@employee_part
POSTHOOK: Output: default@employee_part@employeesalary=3000.0/country=UK
-FAILED: SemanticException [Error 30007]: Invalid partitioning key/value specified in ANALYZE statement
+FAILED: SemanticException [Error 30007]: Invalid partitioning key/value specified in ANALYZE statement : {employeesalary=4000.0, country=Canada}
Modified: hive/branches/cbo/ql/src/test/results/clientpositive/columnstats_partlvl.q.out
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/test/results/clientpositive/columnstats_partlvl.q.out?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
Files hive/branches/cbo/ql/src/test/results/clientpositive/columnstats_partlvl.q.out (original) and hive/branches/cbo/ql/src/test/results/clientpositive/columnstats_partlvl.q.out Tue Jul 1 02:15:23 2014 differ
Modified: hive/branches/cbo/ql/src/test/results/compiler/plan/groupby1.q.xml
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/test/results/compiler/plan/groupby1.q.xml?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/test/results/compiler/plan/groupby1.q.xml (original)
+++ hive/branches/cbo/ql/src/test/results/compiler/plan/groupby1.q.xml Tue Jul 1 02:15:23 2014
@@ -363,7 +363,7 @@
<object class="java.util.ArrayList"/>
</void>
<void property="keyCols">
- <object id="ArrayList0" class="java.util.ArrayList">
+ <object class="java.util.ArrayList">
<void method="add">
<object idref="ExprNodeColumnDesc0"/>
</void>
@@ -423,7 +423,11 @@
</object>
</void>
<void property="partitionCols">
- <object idref="ArrayList0"/>
+ <object class="java.util.ArrayList">
+ <void method="add">
+ <object idref="ExprNodeColumnDesc0"/>
+ </void>
+ </object>
</void>
<void property="tag">
<int>-1</int>
@@ -615,7 +619,7 @@
</object>
</void>
<void property="listGroupingSets">
- <object id="ArrayList1" class="java.util.ArrayList"/>
+ <object id="ArrayList0" class="java.util.ArrayList"/>
</void>
<void property="memoryThreshold">
<float>0.9</float>
@@ -1328,7 +1332,7 @@
</object>
</void>
<void property="listGroupingSets">
- <object idref="ArrayList1"/>
+ <object idref="ArrayList0"/>
</void>
<void property="memoryThreshold">
<float>0.9</float>
Modified: hive/branches/cbo/ql/src/test/results/compiler/plan/groupby4.q.xml
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/test/results/compiler/plan/groupby4.q.xml?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/test/results/compiler/plan/groupby4.q.xml (original)
+++ hive/branches/cbo/ql/src/test/results/compiler/plan/groupby4.q.xml Tue Jul 1 02:15:23 2014
@@ -197,7 +197,7 @@
<object class="java.util.ArrayList"/>
</void>
<void property="keyCols">
- <object id="ArrayList0" class="java.util.ArrayList">
+ <object class="java.util.ArrayList">
<void method="add">
<object idref="ExprNodeColumnDesc0"/>
</void>
@@ -253,7 +253,11 @@
<object class="java.util.ArrayList"/>
</void>
<void property="partitionCols">
- <object idref="ArrayList0"/>
+ <object class="java.util.ArrayList">
+ <void method="add">
+ <object idref="ExprNodeColumnDesc0"/>
+ </void>
+ </object>
</void>
<void property="tag">
<int>-1</int>
@@ -405,7 +409,7 @@
</object>
</void>
<void property="listGroupingSets">
- <object id="ArrayList1" class="java.util.ArrayList"/>
+ <object id="ArrayList0" class="java.util.ArrayList"/>
</void>
<void property="memoryThreshold">
<float>0.9</float>
@@ -1009,7 +1013,7 @@
</object>
</void>
<void property="listGroupingSets">
- <object idref="ArrayList1"/>
+ <object idref="ArrayList0"/>
</void>
<void property="memoryThreshold">
<float>0.9</float>
Modified: hive/branches/cbo/ql/src/test/results/compiler/plan/groupby5.q.xml
URL: http://svn.apache.org/viewvc/hive/branches/cbo/ql/src/test/results/compiler/plan/groupby5.q.xml?rev=1606964&r1=1606963&r2=1606964&view=diff
==============================================================================
--- hive/branches/cbo/ql/src/test/results/compiler/plan/groupby5.q.xml (original)
+++ hive/branches/cbo/ql/src/test/results/compiler/plan/groupby5.q.xml Tue Jul 1 02:15:23 2014
@@ -215,7 +215,7 @@
<object class="java.util.ArrayList"/>
</void>
<void property="keyCols">
- <object id="ArrayList0" class="java.util.ArrayList">
+ <object class="java.util.ArrayList">
<void method="add">
<object idref="ExprNodeColumnDesc0"/>
</void>
@@ -275,7 +275,11 @@
</object>
</void>
<void property="partitionCols">
- <object idref="ArrayList0"/>
+ <object class="java.util.ArrayList">
+ <void method="add">
+ <object idref="ExprNodeColumnDesc0"/>
+ </void>
+ </object>
</void>
<void property="tag">
<int>-1</int>
@@ -467,7 +471,7 @@
</object>
</void>
<void property="listGroupingSets">
- <object id="ArrayList1" class="java.util.ArrayList"/>
+ <object id="ArrayList0" class="java.util.ArrayList"/>
</void>
<void property="memoryThreshold">
<float>0.9</float>
@@ -1203,7 +1207,7 @@
</object>
</void>
<void property="listGroupingSets">
- <object idref="ArrayList1"/>
+ <object idref="ArrayList0"/>
</void>
<void property="memoryThreshold">
<float>0.9</float>