You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by na...@apache.org on 2013/01/30 05:20:41 UTC
svn commit: r1440271 [1/7] - in /hive/trunk: ./
common/src/java/org/apache/hadoop/hive/conf/ conf/
ql/src/java/org/apache/hadoop/hive/ql/exec/
ql/src/java/org/apache/hadoop/hive/ql/lib/
ql/src/java/org/apache/hadoop/hive/ql/metadata/ ql/src/java/org/ap...
Author: namit
Date: Wed Jan 30 04:20:39 2013
New Revision: 1440271
URL: http://svn.apache.org/viewvc?rev=1440271&view=rev
Log:
HIVE-933 Infer bucketing/sorting properties
(Kevin Wilfong via namit)
Added:
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/RuleExactMatch.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingCtx.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java
hive/trunk/ql/src/test/queries/clientnegative/merge_negative_3.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_bucketed_table.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_convert_join.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_dyn_part.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_grouping_operators.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_list_bucket.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_map_operators.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_merge.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_multi_insert.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_num_buckets.q
hive/trunk/ql/src/test/queries/clientpositive/infer_bucket_sort_reducers_power_two.q
hive/trunk/ql/src/test/results/clientnegative/merge_negative_3.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_bucketed_table.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_convert_join.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_dyn_part.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_grouping_operators.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_list_bucket.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_map_operators.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_merge.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_multi_insert.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_num_buckets.q.out
hive/trunk/ql/src/test/results/clientpositive/infer_bucket_sort_reducers_power_two.q.out
Modified:
hive/trunk/build-common.xml
hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
hive/trunk/conf/hive-default.xml.template
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExtractOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ForwardOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapRedTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredWork.java
hive/trunk/ql/src/test/queries/clientpositive/reduce_deduplicate_exclude_gby.q
hive/trunk/ql/src/test/results/clientpositive/ctas.q.out
hive/trunk/ql/src/test/results/compiler/plan/case_sensitivity.q.xml
hive/trunk/ql/src/test/results/compiler/plan/cast1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/groupby6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input20.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input7.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input8.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input9.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_part1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_testsequencefile.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_testxpath.q.xml
hive/trunk/ql/src/test/results/compiler/plan/input_testxpath2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join7.q.xml
hive/trunk/ql/src/test/results/compiler/plan/join8.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample2.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample3.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample5.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/sample7.q.xml
hive/trunk/ql/src/test/results/compiler/plan/subq.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf1.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf4.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf6.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf_case.q.xml
hive/trunk/ql/src/test/results/compiler/plan/udf_when.q.xml
hive/trunk/ql/src/test/results/compiler/plan/union.q.xml
Modified: hive/trunk/build-common.xml
URL: http://svn.apache.org/viewvc/hive/trunk/build-common.xml?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/build-common.xml (original)
+++ hive/trunk/build-common.xml Wed Jan 30 04:20:39 2013
@@ -57,7 +57,7 @@
<property name="test.output" value="true"/>
<property name="test.junit.output.format" value="xml"/>
<property name="test.junit.output.usefile" value="true"/>
- <property name="minimr.query.files" value="list_bucket_dml_10.q,input16_cc.q,scriptfile1.q,bucket4.q,bucketmapjoin6.q,disable_merge_for_bucketing.q,reduce_deduplicate.q,smb_mapjoin_8.q,join1.q,groupby2.q,bucketizedhiveinputformat.q,bucketmapjoin7.q,optrstat_groupby.q,bucket_num_reducers.q,bucket5.q,load_fs2.q,bucket_num_reducers2.q,operatorhook.q"/>
+ <property name="minimr.query.files" value="list_bucket_dml_10.q,input16_cc.q,scriptfile1.q,bucket4.q,bucketmapjoin6.q,disable_merge_for_bucketing.q,reduce_deduplicate.q,smb_mapjoin_8.q,join1.q,groupby2.q,bucketizedhiveinputformat.q,bucketmapjoin7.q,optrstat_groupby.q,bucket_num_reducers.q,bucket5.q,load_fs2.q,bucket_num_reducers2.q,operatorhook.q,infer_bucket_sort_merge.q,infer_bucket_sort_reducers_power_two.q,infer_bucket_sort_dyn_part.q,infer_bucket_sort_bucketed_table.q,infer_bucket_sort_map_operators.q,infer_bucket_sort_num_buckets.q"/>
<property name="minimr.query.negative.files" value="cluster_tasklog_retrieval.q,minimr_broken_pipe.q,mapreduce_stack_trace.q,mapreduce_stack_trace_turnoff.q,mapreduce_stack_trace_hadoop20.q,mapreduce_stack_trace_turnoff_hadoop20.q" />
<property name="test.silent" value="true"/>
<property name="hadoopVersion" value="${hadoop.version.ant-internal}"/>
Modified: hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
URL: http://svn.apache.org/viewvc/hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java (original)
+++ hive/trunk/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java Wed Jan 30 04:20:39 2013
@@ -673,6 +673,17 @@ public class HiveConf extends Configurat
HIVE_MULTI_INSERT_MOVE_TASKS_SHARE_DEPENDENCIES(
"hive.multi.insert.move.tasks.share.dependencies", false),
+ // If this is set, when writing partitions, the metadata will include the bucketing/sorting
+ // properties with which the data was written if any (this will not overwrite the metadata
+ // inherited from the table if the table is bucketed/sorted)
+ HIVE_INFER_BUCKET_SORT("hive.exec.infer.bucket.sort", false),
+ // If this is set, when setting the number of reducers for the map reduce task which writes the
+ // final output files, it will choose a number which is a power of two. The number of reducers
+ // may be set to a power of two, only to be followed by a merge task meaning preventing
+ // anything from being inferred.
+ HIVE_INFER_BUCKET_SORT_NUM_BUCKETS_POWER_TWO(
+ "hive.exec.infer.bucket.sort.num.buckets.power.two", false),
+
/* The following section contains all configurations used for list bucketing feature.*/
/* This is not for clients. but only for block merge task. */
/* This is used by BlockMergeTask to send out flag to RCFileMergeMapper */
Modified: hive/trunk/conf/hive-default.xml.template
URL: http://svn.apache.org/viewvc/hive/trunk/conf/hive-default.xml.template?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/conf/hive-default.xml.template (original)
+++ hive/trunk/conf/hive-default.xml.template Wed Jan 30 04:20:39 2013
@@ -1663,6 +1663,39 @@
</property>
<property>
+ <name>hive.exec.infer.bucket.sort</name>
+ <value>false</value>
+ <description>
+ If this is set, when writing partitions, the metadata will include the bucketing/sorting
+ properties with which the data was written if any (this will not overwrite the metadata
+ inherited from the table if the table is bucketed/sorted)
+ </description>
+</property>
+
+<property>
+ <name>hive.exec.infer.bucket.sort.num.buckets.power.two</name>
+ <value>false</value>
+ <description>
+ If this is set, when setting the number of reducers for the map reduce task which writes the
+ final output files, it will choose a number which is a power of two, unless the user specifies
+ the number of reducers to use using mapred.reduce.tasks. The number of reducers
+ may be set to a power of two, only to be followed by a merge task meaning preventing
+ anything from being inferred.
+ With hive.exec.infer.bucket.sort set to true:
+ Advantages: If this is not set, the number of buckets for partitions will seem arbitrary,
+ which means that the number of mappers used for optimized joins, for example, will
+ be very low. With this set, since the number of buckets used for any partition is
+ a power of two, the number of mappers used for optimized joins will be the least
+ number of buckets used by any partition being joined.
+ Disadvantages: This may mean a much larger or much smaller number of reducers being used in the
+ final map reduce job, e.g. if a job was originally going to take 257 reducers,
+ it will now take 512 reducers, similarly if the max number of reducers is 511,
+ and a job was going to use this many, it will now use 256 reducers.
+
+ </description>
+</property>
+
+<property>
<name>hive.groupby.orderby.position.alias</name>
<value>false</value>
<description>Whether to enable using Column Position Alias in Group By or Order By</description>
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ConditionalTask.java Wed Jan 30 04:20:39 2013
@@ -91,6 +91,12 @@ public class ConditionalTask extends Tas
//recursively remove this task from its children's parent task
tsk.removeFromChildrenTasks();
} else {
+ if (getParentTasks() != null) {
+ // This makes it so that we can go back up the tree later
+ for (Task<? extends Serializable> task : getParentTasks()) {
+ task.addDependentTask(tsk);
+ }
+ }
// resolved task
if (!driverContext.getRunnable().contains(tsk)) {
console.printInfo(tsk.getId() + " is selected by condition resolver.");
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExtractOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExtractOperator.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExtractOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ExtractOperator.java Wed Jan 30 04:20:39 2013
@@ -49,4 +49,16 @@ public class ExtractOperator extends Ope
public OperatorType getType() {
return OperatorType.EXTRACT;
}
+
+ /**
+ * @return the name of the operator
+ */
+ @Override
+ public String getName() {
+ return getOperatorName();
+ }
+
+ static public String getOperatorName() {
+ return "EX";
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ForwardOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ForwardOperator.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ForwardOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ForwardOperator.java Wed Jan 30 04:20:39 2013
@@ -40,4 +40,16 @@ public class ForwardOperator extends Ope
public OperatorType getType() {
return OperatorType.FORWARD;
}
+
+ /**
+ * @return the name of the operator
+ */
+ @Override
+ public String getName() {
+ return getOperatorName();
+ }
+
+ static public String getOperatorName() {
+ return "FOR";
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapRedTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapRedTask.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapRedTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MapRedTask.java Wed Jan 30 04:20:39 2013
@@ -422,6 +422,10 @@ public class MapRedTask extends ExecDriv
* Estimate the number of reducers needed for this job, based on job input,
* and configuration parameters.
*
+ * The output of this method should only be used if the output of this
+ * MapRedTask is not being used to populate a bucketed table and the user
+ * has not specified the number of reducers to use.
+ *
* @return the number of reducers.
*/
private int estimateNumberOfReducers() throws IOException {
@@ -447,6 +451,30 @@ public class MapRedTask extends ExecDriv
int reducers = (int) ((totalInputFileSize + bytesPerReducer - 1) / bytesPerReducer);
reducers = Math.max(1, reducers);
reducers = Math.min(maxReducers, reducers);
+
+ // If this map reduce job writes final data to a table and bucketing is being inferred,
+ // and the user has configured Hive to do this, make sure the number of reducers is a
+ // power of two
+ if (conf.getBoolVar(HiveConf.ConfVars.HIVE_INFER_BUCKET_SORT_NUM_BUCKETS_POWER_TWO) &&
+ work.isFinalMapRed() && !work.getBucketedColsByDirectory().isEmpty()) {
+
+ int reducersLog = (int)(Math.log(reducers) / Math.log(2)) + 1;
+ int reducersPowerTwo = (int)Math.pow(2, reducersLog);
+
+ // If the original number of reducers was a power of two, use that
+ if (reducersPowerTwo / 2 == reducers) {
+ return reducers;
+ } else if (reducersPowerTwo > maxReducers) {
+ // If the next power of two greater than the original number of reducers is greater
+ // than the max number of reducers, use the preceding power of two, which is strictly
+ // less than the original number of reducers and hence the max
+ reducers = reducersPowerTwo / 2;
+ } else {
+ // Otherwise use the smallest power of two greater than the original number of reducers
+ reducers = reducersPowerTwo;
+ }
+ }
+
return reducers;
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java Wed Jan 30 04:20:39 2013
@@ -35,21 +35,28 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
+import org.apache.hadoop.hive.metastore.api.Order;
import org.apache.hadoop.hive.ql.Context;
import org.apache.hadoop.hive.ql.DriverContext;
import org.apache.hadoop.hive.ql.hooks.LineageInfo.DataContainer;
import org.apache.hadoop.hive.ql.hooks.WriteEntity;
import org.apache.hadoop.hive.ql.io.HiveFileFormatUtils;
+import org.apache.hadoop.hive.ql.io.rcfile.merge.BlockMergeTask;
import org.apache.hadoop.hive.ql.lockmgr.HiveLock;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockManager;
import org.apache.hadoop.hive.ql.lockmgr.HiveLockObj;
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.optimizer.physical.BucketingSortingCtx.BucketCol;
+import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.SortCol;
+import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer;
import org.apache.hadoop.hive.ql.plan.DynamicPartitionCtx;
import org.apache.hadoop.hive.ql.plan.LoadFileDesc;
import org.apache.hadoop.hive.ql.plan.LoadMultiFilesDesc;
import org.apache.hadoop.hive.ql.plan.LoadTableDesc;
+import org.apache.hadoop.hive.ql.plan.MapredWork;
import org.apache.hadoop.hive.ql.plan.MoveWork;
import org.apache.hadoop.hive.ql.plan.api.StageType;
import org.apache.hadoop.hive.ql.session.SessionState;
@@ -276,6 +283,48 @@ public class MoveTask extends Task<MoveW
}
} else {
LOG.info("Partition is: " + tbd.getPartitionSpec().toString());
+
+ // Check if the bucketing and/or sorting columns were inferred
+ List<BucketCol> bucketCols = null;
+ List<SortCol> sortCols = null;
+ int numBuckets = -1;
+ Task task = this;
+ String path = tbd.getSourceDir();
+ // Find the first ancestor of this MoveTask which is some form of map reduce task
+ // (Either standard, local, or a merge)
+ while (task.getParentTasks() != null && task.getParentTasks().size() == 1) {
+ task = (Task)task.getParentTasks().get(0);
+ // If it was a merge task or a local map reduce task, nothing can be inferred
+ if (task instanceof BlockMergeTask || task instanceof MapredLocalTask) {
+ break;
+ }
+
+ // If it's a standard map reduce task, check what, if anything, it inferred about
+ // the directory this move task is moving
+ if (task instanceof MapRedTask) {
+ MapredWork work = (MapredWork)task.getWork();
+ bucketCols = work.getBucketedColsByDirectory().get(path);
+ sortCols = work.getSortedColsByDirectory().get(path);
+ numBuckets = work.getNumReduceTasks();
+ if (bucketCols != null || sortCols != null) {
+ // This must be a final map reduce task (the task containing the file sink
+ // operator that writes the final output)
+ assert work.isFinalMapRed();
+ }
+ break;
+ }
+
+ // If it's a move task, get the path the files were moved from, this is what any
+ // preceding map reduce task inferred information about, and moving does not invalidate
+ // those assumptions
+ // This can happen when a conditional merge is added before the final MoveTask, but the
+ // condition for merging is not met, see GenMRFileSink1.
+ if (task instanceof MoveTask) {
+ if (((MoveTask)task).getWork().getLoadFileWork() != null) {
+ path = ((MoveTask)task).getWork().getLoadFileWork().getSourceDir();
+ }
+ }
+ }
// deal with dynamic partitions
DynamicPartitionCtx dpCtx = tbd.getDPCtx();
if (dpCtx != null && dpCtx.getNumDPCols() > 0) { // dynamic partitions
@@ -314,6 +363,10 @@ public class MoveTask extends Task<MoveW
for (LinkedHashMap<String, String> partSpec: dp) {
Partition partn = db.getPartition(table, partSpec, false);
+ if (bucketCols != null || sortCols != null) {
+ updatePartitionBucketSortColumns(table, partn, bucketCols, numBuckets, sortCols);
+ }
+
WriteEntity enty = new WriteEntity(partn, true);
if (work.getOutputs() != null) {
work.getOutputs().add(enty);
@@ -344,6 +397,11 @@ public class MoveTask extends Task<MoveW
tbd.getPartitionSpec(), tbd.getReplace(), tbd.getHoldDDLTime(),
tbd.getInheritTableSpecs(), isSkewedStoredAsDirs(tbd));
Partition partn = db.getPartition(table, tbd.getPartitionSpec(), false);
+
+ if (bucketCols != null || sortCols != null) {
+ updatePartitionBucketSortColumns(table, partn, bucketCols, numBuckets, sortCols);
+ }
+
dc = new DataContainer(table.getTTable(), partn.getTPartition());
// add this partition to post-execution hook
if (work.getOutputs() != null) {
@@ -371,6 +429,80 @@ public class MoveTask extends Task<MoveW
.isSkewedStoredAsDir();
}
+ /**
+ * Alters the bucketing and/or sorting columns of the partition provided they meet some
+ * validation criteria, e.g. the number of buckets match the number of files, and the
+ * columns are not partition columns
+ * @param table
+ * @param partn
+ * @param bucketCols
+ * @param numBuckets
+ * @param sortCols
+ * @throws IOException
+ * @throws InvalidOperationException
+ * @throws HiveException
+ */
+ private void updatePartitionBucketSortColumns(Table table, Partition partn,
+ List<BucketCol> bucketCols, int numBuckets, List<SortCol> sortCols)
+ throws IOException, InvalidOperationException, HiveException {
+
+ boolean updateBucketCols = false;
+ if (bucketCols != null) {
+ FileSystem fileSys = partn.getPartitionPath().getFileSystem(conf);
+ FileStatus[] fileStatus = Utilities.getFileStatusRecurse(
+ partn.getPartitionPath(), 1, fileSys);
+ // Verify the number of buckets equals the number of files
+ // This will not hold for dynamic partitions where not every reducer produced a file for
+ // those partitions. In this case the table is not bucketed as Hive requires a files for
+ // each bucket.
+ if (fileStatus.length == numBuckets) {
+ List<String> newBucketCols = new ArrayList<String>();
+ updateBucketCols = true;
+ for (BucketCol bucketCol : bucketCols) {
+ if (bucketCol.getIndexes().get(0) < partn.getCols().size()) {
+ newBucketCols.add(partn.getCols().get(
+ bucketCol.getIndexes().get(0)).getName());
+ } else {
+ // If the table is bucketed on a partition column, not valid for bucketing
+ updateBucketCols = false;
+ break;
+ }
+ }
+ if (updateBucketCols) {
+ partn.getBucketCols().clear();
+ partn.getBucketCols().addAll(newBucketCols);
+ partn.getTPartition().getSd().setNumBuckets(numBuckets);
+ }
+ }
+ }
+
+ boolean updateSortCols = false;
+ if (sortCols != null) {
+ List<Order> newSortCols = new ArrayList<Order>();
+ updateSortCols = true;
+ for (SortCol sortCol : sortCols) {
+ if (sortCol.getIndexes().get(0) < partn.getCols().size()) {
+ newSortCols.add(new Order(
+ partn.getCols().get(sortCol.getIndexes().get(0)).getName(),
+ sortCol.getSortOrder() == '+' ? BaseSemanticAnalyzer.HIVE_COLUMN_ORDER_ASC :
+ BaseSemanticAnalyzer.HIVE_COLUMN_ORDER_DESC));
+ } else {
+ // If the table is sorted on a partition column, not valid for sorting
+ updateSortCols = false;
+ break;
+ }
+ }
+ if (updateSortCols) {
+ partn.getSortCols().clear();
+ partn.getSortCols().addAll(newSortCols);
+ }
+ }
+
+ if (updateBucketCols || updateSortCols) {
+ db.alterPartition(table.getDbName(), table.getTableName(), partn);
+ }
+ }
+
/*
* Does the move task involve moving to a local file system
*/
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/RuleExactMatch.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/RuleExactMatch.java?rev=1440271&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/RuleExactMatch.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/lib/RuleExactMatch.java Wed Jan 30 04:20:39 2013
@@ -0,0 +1,85 @@
+/**
+ * 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.lib;
+
+import java.util.Stack;
+
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+
+/**
+ * Implentation of the Rule interface for Nodes Used in Node dispatching to dispatch
+ * process/visitor functions for Nodes. The cost method returns 1 if there is an exact
+ * match between the expression and the stack, otherwise -1.
+ */
+public class RuleExactMatch implements Rule {
+
+ private final String ruleName;
+ private final String pattern;
+
+ /**
+ * The rule specified as operator names separated by % symbols, the left side represents the
+ * bottom of the stack.
+ *
+ * E.g. TS%FIL%RS -> means
+ * TableScan Node followed by Filter followed by ReduceSink in the tree, or, in terms of the
+ * stack, ReduceSink on top followed by Filter followed by TableScan
+ *
+ * @param ruleName
+ * name of the rule
+ * @param regExp
+ * string specification of the rule
+ **/
+ public RuleExactMatch(String ruleName, String pattern) {
+ this.ruleName = ruleName;
+ this.pattern = pattern;
+ }
+
+ /**
+ * This function returns the cost of the rule for the specified stack. Returns 1 if there is
+ * an exact match with the entire stack, otherwise -1
+ *
+ * If any proper substack of the stack matches it will return -1. It only returns 1 if the
+ * entire stack matches the rule exactly.
+ *
+ * @param stack
+ * Node stack encountered so far
+ * @return cost of the function
+ * @throws SemanticException
+ */
+ public int cost(Stack<Node> stack) throws SemanticException {
+ int numElems = (stack != null ? stack.size() : 0);
+ String name = new String();
+ for (int pos = numElems - 1; pos >= 0; pos--) {
+ name = stack.get(pos).getName() + "%" + name;
+ }
+
+ if (pattern.equals(name)) {
+ return 1;
+ }
+
+ return -1;
+ }
+
+ /**
+ * @return the name of the Node
+ **/
+ public String getName() {
+ return ruleName;
+ }
+}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java Wed Jan 30 04:20:39 2013
@@ -409,7 +409,7 @@ public class Hive {
}
/**
- * Updates the existing table metadata with the new metadata.
+ * Updates the existing partition metadata with the new metadata.
*
* @param tblName
* name of the existing table
@@ -422,13 +422,30 @@ public class Hive {
public void alterPartition(String tblName, Partition newPart)
throws InvalidOperationException, HiveException {
Table t = newTable(tblName);
+ alterPartition(t.getDbName(), t.getTableName(), newPart);
+ }
+
+ /**
+ * Updates the existing partition metadata with the new metadata.
+ *
+ * @param dbName
+ * name of the exiting table's database
+ * @param tblName
+ * name of the existing table
+ * @param newPart
+ * new partition
+ * @throws InvalidOperationException
+ * if the changes in metadata is not acceptable
+ * @throws TException
+ */
+ public void alterPartition(String dbName, String tblName, Partition newPart)
+ throws InvalidOperationException, HiveException {
try {
// Remove the DDL time so that it gets refreshed
if (newPart.getParameters() != null) {
newPart.getParameters().remove(hive_metastoreConstants.DDL_TIME);
}
- getMSC().alter_partition(t.getDbName(), t.getTableName(),
- newPart.getTPartition());
+ getMSC().alter_partition(dbName, tblName, newPart.getTPartition());
} catch (MetaException e) {
throw new HiveException("Unable to alter partition.", e);
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/ColumnPrunerProcFactory.java Wed Jan 30 04:20:39 2013
@@ -548,6 +548,18 @@ public final class ColumnPrunerProcFacto
.getValueCols();
ArrayList<String> newOutputColNames = new ArrayList<String>();
java.util.ArrayList<ExprNodeDesc> newValueEval = new ArrayList<ExprNodeDesc>();
+ // ReduceSinkOperators that precede GroupByOperators have the keys in the schema in addition
+ // to the values. These are not pruned.
+ List<ColumnInfo> oldSchema = oldRR.getRowSchema().getSignature();
+ for (ColumnInfo colInfo : oldSchema) {
+ if (colInfo.getInternalName().startsWith(Utilities.ReduceField.KEY.toString() + ".")) {
+ String[] nm = oldRR.reverseLookup(colInfo.getInternalName());
+ newRR.put(nm[0], nm[1], colInfo);
+ sig.add(colInfo);
+ } else {
+ break;
+ }
+ }
for (int i = 0; i < retainFlags.length; i++) {
if (retainFlags[i]) {
newValueEval.add(originalValueEval.get(i));
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRFileSink1.java Wed Jan 30 04:20:39 2013
@@ -97,6 +97,9 @@ public class GenMRFileSink1 implements N
parseCtx.getQB().getParseInfo().isInsertToTable();
HiveConf hconf = parseCtx.getConf();
+ // Mark this task as a final map reduce task (ignoring the optional merge task)
+ ((MapredWork)currTask.getWork()).setFinalMapRed(true);
+
// If this file sink desc has been processed due to a linked file sink desc,
// use that task
Map<FileSinkDesc, Task<? extends Serializable>> fileSinkDescs = ctx.getLinkedFileDescTasks();
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingCtx.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingCtx.java?rev=1440271&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingCtx.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingCtx.java Wed Jan 30 04:20:39 2013
@@ -0,0 +1,230 @@
+/**
+ * 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.physical;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+
+/**
+ * This class contains the bucketing sorting context that is passed
+ * while walking the operator tree in inferring bucket/sort columns. The context
+ * contains the mappings from operators and files to the columns their output is
+ * bucketed/sorted on.
+ */
+public class BucketingSortingCtx implements NodeProcessorCtx {
+
+ // A mapping from an operator to the columns by which it's output is bucketed
+ Map<Operator<? extends OperatorDesc>, List<BucketCol>> bucketedColsByOp;
+ // A mapping from a directory which a FileSinkOperator writes into to the columns by which that
+ // output is bucketed
+ Map<String, List<BucketCol>> bucketedColsByDirectory;
+
+ // A mapping from an operator to the columns by which it's output is sorted
+ Map<Operator<? extends OperatorDesc>, List<SortCol>> sortedColsByOp;
+ // A mapping from a directory which a FileSinkOperator writes into to the columns by which that
+ // output is sorted
+ Map<String, List<SortCol>> sortedColsByDirectory;
+
+ public BucketingSortingCtx() {
+ this.bucketedColsByOp = new HashMap<Operator<? extends OperatorDesc>, List<BucketCol>>();
+ this.bucketedColsByDirectory = new HashMap<String, List<BucketCol>>();
+ this.sortedColsByOp = new HashMap<Operator<? extends OperatorDesc>, List<SortCol>>();
+ this.sortedColsByDirectory = new HashMap<String, List<SortCol>>();
+ }
+
+
+ public List<BucketCol> getBucketedCols(Operator<? extends OperatorDesc> op) {
+ return bucketedColsByOp.get(op);
+ }
+
+
+ public void setBucketedCols(Operator<? extends OperatorDesc> op, List<BucketCol> bucketCols) {
+ this.bucketedColsByOp.put(op, bucketCols);
+ }
+
+ public Map<String, List<BucketCol>> getBucketedColsByDirectory() {
+ return bucketedColsByDirectory;
+ }
+
+
+ public void setBucketedColsByDirectory(Map<String, List<BucketCol>> bucketedColsByDirectory) {
+ this.bucketedColsByDirectory = bucketedColsByDirectory;
+ }
+
+
+ public List<SortCol> getSortedCols(Operator<? extends OperatorDesc> op) {
+ return sortedColsByOp.get(op);
+ }
+
+
+ public void setSortedCols(Operator<? extends OperatorDesc> op, List<SortCol> sortedCols) {
+ this.sortedColsByOp.put(op, sortedCols);
+ }
+
+ public Map<String, List<SortCol>> getSortedColsByDirectory() {
+ return sortedColsByDirectory;
+ }
+
+
+ public void setSortedColsByDirectory(Map<String, List<SortCol>> sortedColsByDirectory) {
+ this.sortedColsByDirectory = sortedColsByDirectory;
+ }
+
+ /**
+ *
+ * BucketSortCol.
+ *
+ * Classes that implement this interface provide a way to store information about equivalent
+ * columns as their names and indexes in the schema change going into and out of operators. The
+ * definition of equivalent columns is up to the class which uses these classes, e.g.
+ * BucketingSortingOpProcFactory. For example, two columns are equivalent if they
+ * contain exactly the same data. Though, it's possible that two columns contain exactly the
+ * same data and are not known to be equivalent.
+ *
+ * E.g. SELECT key a, key b FROM (SELECT key, count(*) c FROM src GROUP BY key) s;
+ * In this case, assuming this is done in a single map reduce job with the group by operator
+ * processed in the reducer, the data coming out of the group by operator will be bucketed
+ * by key, which would be at index 0 in the schema, after the outer select operator, the output
+ * can be viewed as bucketed by either the column with alias a or the column with alias b. To
+ * represent this, there could be a single BucketSortCol implementation instance whose names
+ * include both a and b, and whose indexes include both 0 and 1.
+ *
+ * Implementations of this interface should maintain the restriction that the alias
+ * getNames().get(i) should have index getIndexes().get(i) in the schema.
+ */
+ public static interface BucketSortCol {
+ // Get a list of aliases for the same column
+ public List<String> getNames();
+
+ // Get a list of indexes for which the columns in the schema are the same
+ public List<Integer> getIndexes();
+
+ // Add an alternative alias for the column this instance represents, and its index in the
+ // schema.
+ public void addAlias(String name, Integer index);
+ }
+
+ /**
+ *
+ * BucketCol.
+ *
+ * An implementation of BucketSortCol which contains known aliases/indexes of equivalent columns
+ * which data is determined to be bucketed on.
+ */
+ public static final class BucketCol implements BucketSortCol, Serializable {
+ private static final long serialVersionUID = 1L;
+ // Equivalent aliases for the column
+ private final List<String> names = new ArrayList<String>();
+ // Indexes of those equivalent columns
+ private final List<Integer> indexes = new ArrayList<Integer>();
+
+ public BucketCol(String name, int index) {
+ addAlias(name, index);
+ }
+
+ public BucketCol() {
+
+ }
+
+ @Override
+ public List<String> getNames() {
+ return names;
+ }
+
+ @Override
+ public List<Integer> getIndexes() {
+ return indexes;
+ }
+
+ @Override
+ public void addAlias(String name, Integer index) {
+ names.add(name);
+ indexes.add(index);
+ }
+
+ @Override
+ // Chooses a representative alias and index to use as the String, the first is used because
+ // it is set in the constructor
+ public String toString() {
+ return "name: " + names.get(0) + " index: " + indexes.get(0);
+ }
+ }
+
+ /**
+ *
+ * SortCol.
+ *
+ * An implementation of BucketSortCol which contains known aliases/indexes of equivalent columns
+ * which data is determined to be sorted on. Unlike aliases, and indexes the sort order is known
+ * to be constant for all equivalent columns.
+ */
+ public static final class SortCol implements BucketSortCol, Serializable {
+ private static final long serialVersionUID = 1L;
+ // Equivalent aliases for the column
+ private final List<String> names = new ArrayList<String>();
+ // Indexes of those equivalent columns
+ private final List<Integer> indexes = new ArrayList<Integer>();
+ // Sort order (+|-)
+ private final char sortOrder;
+
+ public SortCol(String name, int index, char sortOrder) {
+ this(sortOrder);
+ addAlias(name, index);
+ }
+
+ public SortCol(char sortOrder) {
+ this.sortOrder = sortOrder;
+ }
+
+
+ @Override
+ public List<String> getNames() {
+ return names;
+ }
+
+ @Override
+ public List<Integer> getIndexes() {
+ return indexes;
+ }
+
+ @Override
+ public void addAlias(String name, Integer index) {
+ names.add(name);
+ indexes.add(index);
+ }
+
+ public char getSortOrder() {
+ return sortOrder;
+ }
+
+ @Override
+ // Chooses a representative alias, index, and order to use as the String, the first is used
+ // because it is set in the constructor
+ public String toString() {
+ return "name: " + names.get(0) + " index: " + indexes.get(0) + " order: " + sortOrder;
+ }
+ }
+}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java?rev=1440271&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingInferenceOptimizer.java Wed Jan 30 04:20:39 2013
@@ -0,0 +1,150 @@
+/**
+ * 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.physical;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.hadoop.hive.ql.exec.ExecDriver;
+import org.apache.hadoop.hive.ql.exec.ExtractOperator;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.exec.FilterOperator;
+import org.apache.hadoop.hive.ql.exec.ForwardOperator;
+import org.apache.hadoop.hive.ql.exec.GroupByOperator;
+import org.apache.hadoop.hive.ql.exec.JoinOperator;
+import org.apache.hadoop.hive.ql.exec.LateralViewForwardOperator;
+import org.apache.hadoop.hive.ql.exec.LateralViewJoinOperator;
+import org.apache.hadoop.hive.ql.exec.LimitOperator;
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.SelectOperator;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
+import org.apache.hadoop.hive.ql.lib.Dispatcher;
+import org.apache.hadoop.hive.ql.lib.GraphWalker;
+import org.apache.hadoop.hive.ql.lib.Node;
+import org.apache.hadoop.hive.ql.lib.NodeProcessor;
+import org.apache.hadoop.hive.ql.lib.PreOrderWalker;
+import org.apache.hadoop.hive.ql.lib.Rule;
+import org.apache.hadoop.hive.ql.lib.RuleExactMatch;
+import org.apache.hadoop.hive.ql.lib.RuleRegExp;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+
+/**
+ *
+ * BucketingSortingInferenceOptimizer.
+ *
+ * For each map reduce task, attmepts to infer bucketing and sorting metadata for the outputs.
+ *
+ * Currently only map reduce tasks which produce final output have there output metadata inferred,
+ * but it can be extended to intermediate tasks as well.
+ *
+ * This should be run as the last physical optimizer, as other physical optimizers may invalidate
+ * the inferences made. If a physical optimizer depends on the results and is designed to
+ * carefully maintain these inferences, it may follow this one.
+ */
+public class BucketingSortingInferenceOptimizer implements PhysicalPlanResolver {
+
+ @Override
+ public PhysicalContext resolve(PhysicalContext pctx) throws SemanticException {
+ inferBucketingSorting(Utilities.getMRTasks(pctx.rootTasks));
+ return pctx;
+ }
+
+ /**
+ * For each map reduce task, if it has a reducer, infer whether or not the final output of the
+ * reducer is bucketed and/or sorted
+ *
+ * @param mapRedTasks
+ * @throws SemanticException
+ */
+ private void inferBucketingSorting(List<ExecDriver> mapRedTasks) throws SemanticException {
+ for (ExecDriver mapRedTask : mapRedTasks) {
+
+ // For now this only is used to determine the bucketing/sorting of outputs, in the future
+ // this can be removed to optimize the query plan based on the bucketing/sorting properties
+ // of the outputs of intermediate map reduce jobs.
+ if (!mapRedTask.getWork().isFinalMapRed()) {
+ continue;
+ }
+
+ Operator<? extends OperatorDesc> reducer = mapRedTask.getWork().getReducer();
+ if (reducer == null) {
+ continue;
+ }
+
+ BucketingSortingCtx bCtx = new BucketingSortingCtx();
+
+ // RuleRegExp rules are used to match operators anywhere in the tree
+ // RuleExactMatch rules are used to specify exactly what the tree should look like
+ // In particular, this guarantees that the first operator is the reducer
+ // (and its parent(s) are ReduceSinkOperators) since it begins walking the tree from
+ // the reducer.
+ Map<Rule, NodeProcessor> opRules = new LinkedHashMap<Rule, NodeProcessor>();
+ opRules.put(new RuleRegExp("R1", SelectOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getSelProc());
+ // Matches only GroupByOpeartors which are reducers, rather than map group by operators,
+ // or multi group by optimization specific operators
+ opRules.put(new RuleExactMatch("R2", GroupByOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getGroupByProc());
+ // Matches only JoinOperators which are reducers, rather than map joins, SMB map joins, etc.
+ opRules.put(new RuleExactMatch("R3", JoinOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getJoinProc());
+ opRules.put(new RuleRegExp("R5", FileSinkOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getFileSinkProc());
+ // Matches only ExtractOperators which are reducers
+ opRules.put(new RuleExactMatch("R6", ExtractOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getExtractProc());
+ opRules.put(new RuleRegExp("R7", FilterOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getFilterProc());
+ opRules.put(new RuleRegExp("R8", LimitOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getLimitProc());
+ opRules.put(new RuleRegExp("R9", LateralViewForwardOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getLateralViewForwardProc());
+ opRules.put(new RuleRegExp("R10", LateralViewJoinOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getLateralViewJoinProc());
+ // Matches only ForwardOperators which are preceded by some other operator in the tree,
+ // in particular it can't be a reducer (and hence cannot be one of the ForwardOperators
+ // added by the multi group by optimization)
+ opRules.put(new RuleRegExp("R11", ".+" + ForwardOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getForwardProc());
+ // Matches only ForwardOperators which are reducers and are followed by GroupByOperators
+ // (specific to the multi group by optimization)
+ opRules.put(new RuleExactMatch("R12", ForwardOperator.getOperatorName() + "%" +
+ GroupByOperator.getOperatorName() + "%"),
+ BucketingSortingOpProcFactory.getMultiGroupByProc());
+
+ // The dispatcher fires the processor corresponding to the closest matching rule and passes
+ // the context along
+ Dispatcher disp = new DefaultRuleDispatcher(BucketingSortingOpProcFactory.getDefaultProc(),
+ opRules, bCtx);
+ GraphWalker ogw = new PreOrderWalker(disp);
+
+ // Create a list of topop nodes
+ ArrayList<Node> topNodes = new ArrayList<Node>();
+ topNodes.add(reducer);
+ ogw.startWalking(topNodes, null);
+
+ mapRedTask.getWork().getBucketedColsByDirectory().putAll(bCtx.getBucketedColsByDirectory());
+ mapRedTask.getWork().getSortedColsByDirectory().putAll(bCtx.getSortedColsByDirectory());
+ }
+ }
+}
Added: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java?rev=1440271&view=auto
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java (added)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java Wed Jan 30 04:20:39 2013
@@ -0,0 +1,785 @@
+/**
+ * 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.physical;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Stack;
+
+import org.apache.hadoop.hive.ql.exec.ColumnInfo;
+import org.apache.hadoop.hive.ql.exec.ExtractOperator;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.exec.ForwardOperator;
+import org.apache.hadoop.hive.ql.exec.GroupByOperator;
+import org.apache.hadoop.hive.ql.exec.JoinOperator;
+import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
+import org.apache.hadoop.hive.ql.exec.SelectOperator;
+import org.apache.hadoop.hive.ql.lib.Node;
+import org.apache.hadoop.hive.ql.lib.NodeProcessor;
+import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.lib.Utils;
+import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.BucketCol;
+import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.BucketSortCol;
+import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.SortCol;
+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.ExprNodeDesc.ExprNodeDescEqualityWrapper;
+import org.apache.hadoop.hive.ql.plan.OperatorDesc;
+
+/**
+ * Operator factory for the rule processors for inferring bucketing/sorting columns.
+ */
+public class BucketingSortingOpProcFactory {
+
+ public static class DefaultInferrer implements NodeProcessor {
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Infers bucket/sort columns for operators which simply forward rows from the parent
+ * E.g. Forward operators and SELECT *
+ * @param op
+ * @param bctx
+ * @param parent
+ * @throws SemanticException
+ */
+ private static void processForward(Operator<? extends OperatorDesc> op, BucketingSortingCtx bctx,
+ Operator<? extends OperatorDesc> parent) throws SemanticException {
+
+ List<BucketCol> bucketCols = bctx.getBucketedCols(parent);
+ List<SortCol> sortCols = bctx.getSortedCols(parent);
+ List<ColumnInfo> colInfos = op.getSchema().getSignature();
+
+ if (bucketCols == null && sortCols == null) {
+ return;
+ }
+
+ List<BucketCol> newBucketCols;
+ List<SortCol> newSortCols;
+
+ if (bucketCols == null) {
+ newBucketCols = null;
+ } else {
+ newBucketCols = getNewBucketCols(bucketCols, colInfos);
+ }
+
+ if (sortCols == null) {
+ newSortCols = null;
+ } else {
+ newSortCols = getNewSortCols(sortCols, colInfos);
+ }
+
+ bctx.setBucketedCols(op, newBucketCols);
+ bctx.setSortedCols(op, newSortCols);
+ }
+
+ /**
+ * Returns the parent operator in the walk path to the current operator.
+ *
+ * @param stack The stack encoding the path.
+ *
+ * @return Operator The parent operator in the current path.
+ */
+ @SuppressWarnings("unchecked")
+ protected static Operator<? extends OperatorDesc> getParent(Stack<Node> stack) {
+ return (Operator<? extends OperatorDesc>)Utils.getNthAncestor(stack, 1);
+ }
+
+ /**
+ * Processor for Join Operator.
+ *
+ * This handles common joins, the tree should look like
+ * ReducSinkOperator
+ * \
+ * .... --- JoinOperator
+ * /
+ * ReduceSink Operator
+ *
+ */
+ public static class JoinInferrer extends DefaultInferrer implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ BucketingSortingCtx bctx = (BucketingSortingCtx)procCtx;
+ JoinOperator jop = (JoinOperator)nd;
+ List<ColumnInfo> colInfos = jop.getSchema().getSignature();
+ Byte[] order = jop.getConf().getTagOrder();
+
+ BucketCol[] newBucketCols = null;
+ SortCol[] newSortCols = null;
+
+ for (int i = 0; i < jop.getParentOperators().size(); i++) {
+
+ Operator<? extends OperatorDesc> parent = jop.getParentOperators().get(i);
+
+ // The caller of this method should guarantee this
+ assert(parent instanceof ReduceSinkOperator);
+
+ ReduceSinkOperator rop = (ReduceSinkOperator)jop.getParentOperators().get(i);
+
+ String sortOrder = rop.getConf().getOrder();
+ List<BucketCol> bucketCols = new ArrayList<BucketCol>();
+ List<SortCol> sortCols = new ArrayList<SortCol>();
+ // Go through the Reduce keys and find the matching column(s) in the reduce values
+ for (int keyIndex = 0; keyIndex < rop.getConf().getKeyCols().size(); keyIndex++) {
+ for (int valueIndex = 0; valueIndex < rop.getConf().getValueCols().size();
+ valueIndex++) {
+
+ if (new ExprNodeDescEqualityWrapper(rop.getConf().getValueCols().get(valueIndex)).
+ equals(new ExprNodeDescEqualityWrapper(rop.getConf().getKeyCols().get(
+ keyIndex)))) {
+
+ String colName = rop.getSchema().getSignature().get(valueIndex).getInternalName();
+ bucketCols.add(new BucketCol(colName, keyIndex));
+ sortCols.add(new SortCol(colName, keyIndex, sortOrder.charAt(keyIndex)));
+ break;
+ }
+ }
+ }
+
+ if (bucketCols.isEmpty()) {
+ assert(sortCols.isEmpty());
+ continue;
+ }
+
+ if (newBucketCols == null) {
+ assert(newSortCols == null);
+ // The number of join keys is equal to the number of keys in every reducer, although
+ // not every key may map to a value in the reducer
+ newBucketCols = new BucketCol[rop.getConf().getKeyCols().size()];
+ newSortCols = new SortCol[rop.getConf().getKeyCols().size()];
+ } else {
+ assert(newSortCols != null);
+ }
+
+ byte tag = (byte)rop.getConf().getTag();
+ List<ExprNodeDesc> exprs = jop.getConf().getExprs().get(tag);
+
+ int colInfosOffset = 0;
+ int orderValue = order[tag];
+ // Columns are output from the join from the different reduce sinks in the order of their
+ // offsets
+ for (byte orderIndex = 0; orderIndex < order.length; orderIndex++) {
+ if (order[orderIndex] < orderValue) {
+ colInfosOffset += jop.getConf().getExprs().get(orderIndex).size();
+ }
+ }
+
+ findBucketingSortingColumns(exprs, colInfos, bucketCols, sortCols, newBucketCols,
+ newSortCols, colInfosOffset);
+
+ }
+
+ setBucketingColsIfComplete(bctx, jop, newBucketCols);
+
+ setSortingColsIfComplete(bctx, jop, newSortCols);
+
+ return null;
+ }
+
+ }
+
+ /**
+ * If the list of output bucket columns has been populated and every column has at least
+ * one representative in the output they can be inferred
+ *
+ * @param bctx - BucketingSortingCtx containing inferred columns
+ * @param op - The operator we are inferring information about the output of
+ * @param newBucketCols - An array of columns on which the output is bucketed, e.g. as output by
+ * the method findBucketingSortingColumns
+ */
+ private static void setBucketingColsIfComplete(BucketingSortingCtx bctx,
+ Operator<? extends OperatorDesc> op, BucketCol[] newBucketCols) {
+
+ if (newBucketCols != null) {
+ List<BucketCol> newBucketColList = Arrays.asList(newBucketCols);
+ // If newBucketColList had a null value it means that at least one of the input bucket
+ // columns did not have a representative found in the output columns, so assume the data
+ // is no longer bucketed
+ if (!newBucketColList.contains(null)) {
+ bctx.setBucketedCols(op, newBucketColList);
+ }
+ }
+ }
+
+ /**
+ * If the list of output sort columns has been populated and every column has at least
+ * one representative in the output they can be inferred
+ *
+ * @param bctx - BucketingSortingCtx containing inferred columns
+ * @param op - The operator we are inferring information about the output of
+ * @param newSortCols - An array of columns on which the output is sorted, e.g. as output by
+ * the method findBucketingSortingColumns
+ */
+ private static void setSortingColsIfComplete(BucketingSortingCtx bctx,
+ Operator<? extends OperatorDesc> op, SortCol[] newSortCols) {
+
+ if (newSortCols != null) {
+ List<SortCol> newSortColList = Arrays.asList(newSortCols);
+ // If newSortColList had a null value it means that at least one of the input sort
+ // columns did not have a representative found in the output columns, so assume the data
+ // is no longer sorted
+ if (!newSortColList.contains(null)) {
+ bctx.setSortedCols(op, newSortColList);
+ }
+ }
+ }
+
+ private static void findBucketingSortingColumns(List<ExprNodeDesc> exprs,
+ List<ColumnInfo> colInfos, List<BucketCol> bucketCols, List<SortCol> sortCols,
+ BucketCol[] newBucketCols, SortCol[] newSortCols) {
+ findBucketingSortingColumns(exprs, colInfos, bucketCols, sortCols, newBucketCols,
+ newSortCols, 0);
+ }
+
+ /**
+ * For each expression, check if it represents a column known to be bucketed/sorted.
+ *
+ * The methods setBucketingColsIfComplete and setSortingColsIfComplete should be used to assign
+ * the values of newBucketCols and newSortCols as the bucketing/sorting columns of this operator
+ * because these arrays may contain nulls indicating that the output of this operator is not
+ * bucketed/sorted.
+ *
+ * @param exprs - list of expression
+ * @param colInfos - list of column infos
+ * @param bucketCols - list of bucketed columns from the input
+ * @param sortCols - list of sorted columns from the input
+ * @param newBucketCols - an array of bucket columns which should be the same length as
+ * bucketCols, updated such that the bucketed column(s) at index i in bucketCols became
+ * the bucketed column(s) at index i of newBucketCols in the output
+ * @param newSortCols - an array of sort columns which should be the same length as
+ * sortCols, updated such that the sorted column(s) at index i in sortCols became
+ * the sorted column(s) at index i of sortCols in the output
+ * @param colInfosOffset - the expressions are known to be represented by column infos
+ * beginning at this index
+ */
+ private static void findBucketingSortingColumns(List<ExprNodeDesc> exprs,
+ List<ColumnInfo> colInfos, List<BucketCol> bucketCols, List<SortCol> sortCols,
+ BucketCol[] newBucketCols, SortCol[] newSortCols, int colInfosOffset) {
+ for(int cnt = 0; cnt < exprs.size(); cnt++) {
+ ExprNodeDesc expr = exprs.get(cnt);
+
+ // Only columns can be sorted/bucketed, in particular applying a function to a column
+ // voids any assumptions
+ if (!(expr instanceof ExprNodeColumnDesc)) {
+ continue;
+ }
+
+ ExprNodeColumnDesc columnExpr = (ExprNodeColumnDesc)expr;
+
+ int colInfosIndex = cnt + colInfosOffset;
+
+ if (newBucketCols != null) {
+ int bucketIndex = indexOfColName(bucketCols, columnExpr.getColumn());
+ if (bucketIndex != -1) {
+ if (newBucketCols[bucketIndex] == null) {
+ newBucketCols[bucketIndex] = new BucketCol();
+ }
+ newBucketCols[bucketIndex].addAlias(
+ colInfos.get(colInfosIndex).getInternalName(), colInfosIndex);
+ }
+ }
+
+ if (newSortCols != null) {
+ int sortIndex = indexOfColName(sortCols, columnExpr.getColumn());
+ if (sortIndex != -1) {
+ if (newSortCols[sortIndex] == null) {
+ newSortCols[sortIndex] = new SortCol(sortCols.get(sortIndex).getSortOrder());
+ }
+ newSortCols[sortIndex].addAlias(
+ colInfos.get(colInfosIndex).getInternalName(), colInfosIndex);
+ }
+ }
+ }
+ }
+
+ /**
+ * Processor for Select operator.
+ */
+ public static class SelectInferrer extends DefaultInferrer implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ BucketingSortingCtx bctx = (BucketingSortingCtx)procCtx;
+ SelectOperator sop = (SelectOperator)nd;
+
+ Operator<? extends OperatorDesc> parent = getParent(stack);
+
+ // if this is a selStarNoCompute then this select operator
+ // is treated like a default operator, so just call the super classes
+ // process method.
+ if (sop.getConf().isSelStarNoCompute()) {
+ processForward(sop, bctx, parent);
+ return null;
+ }
+
+ List<BucketCol> bucketCols = bctx.getBucketedCols(parent);
+ List<SortCol> sortCols = bctx.getSortedCols(parent);
+ List<ColumnInfo> colInfos = sop.getSchema().getSignature();
+
+ if (bucketCols == null && sortCols == null) {
+ return null;
+ }
+
+ BucketCol[] newBucketCols = null;
+ SortCol[] newSortCols = null;
+ if (bucketCols != null) {
+ newBucketCols = new BucketCol[bucketCols.size()];
+ }
+ if (sortCols != null) {
+ newSortCols = new SortCol[sortCols.size()];
+ }
+
+ findBucketingSortingColumns(sop.getConf().getColList(), colInfos, bucketCols, sortCols,
+ newBucketCols, newSortCols);
+
+ setBucketingColsIfComplete(bctx, sop, newBucketCols);
+
+ setSortingColsIfComplete(bctx, sop, newSortCols);
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Find the BucketSortCol which has colName as one of its aliases. Returns the index of that
+ * BucketSortCol, or -1 if none exist
+ * @param bucketSortCols
+ * @param colName
+ * @return
+ */
+ private static int indexOfColName(List<? extends BucketSortCol> bucketSortCols, String colName) {
+ for (int index = 0; index < bucketSortCols.size(); index++) {
+ BucketSortCol bucketSortCol = bucketSortCols.get(index);
+ if (bucketSortCol.getNames().indexOf(colName) != -1) {
+ return index;
+ }
+ }
+
+ return -1;
+ }
+
+ /**
+ * This is used to construct new lists of bucketed columns where the order of the columns
+ * hasn't changed, only possibly the name
+ * @param bucketCols - input bucketed columns
+ * @param colInfos - List of column infos
+ * @return output bucketed columns
+ */
+ private static List<BucketCol> getNewBucketCols(List<BucketCol> bucketCols,
+ List<ColumnInfo> colInfos) {
+
+ List<BucketCol> newBucketCols = new ArrayList<BucketCol>(bucketCols.size());
+ for (int i = 0; i < bucketCols.size(); i++) {
+ BucketCol bucketCol = new BucketCol();
+ for (Integer index : bucketCols.get(i).getIndexes()) {
+ // The only time this condition should be false is in the case of dynamic partitioning
+ // where the data is bucketed on a dynamic partitioning column and the FileSinkOperator is
+ // being processed. In this case, the dynamic partition column will not appear in
+ // colInfos, and due to the limitations of dynamic partitioning, they will appear at the
+ // end of the input schema. Since the order of the columns hasn't changed, and no new
+ // columns have been added/removed, it is safe to assume that these will have indexes
+ // greater than or equal to colInfos.size().
+ if (index < colInfos.size()) {
+ bucketCol.addAlias(colInfos.get(index).getInternalName(), index);
+ } else {
+ return null;
+ }
+ }
+ newBucketCols.add(bucketCol);
+ }
+ return newBucketCols;
+ }
+
+ /**
+ * This is used to construct new lists of sorted columns where the order of the columns
+ * hasn't changed, only possibly the name
+ * @param bucketCols - input sorted columns
+ * @param colInfos - List of column infos
+ * @return output sorted columns
+ */
+ private static List<SortCol> getNewSortCols(List<SortCol> sortCols, List<ColumnInfo> colInfos) {
+ List<SortCol> newSortCols = new ArrayList<SortCol>(sortCols.size());
+ for (int i = 0; i < sortCols.size(); i++) {
+ SortCol sortCol = new SortCol(sortCols.get(i).getSortOrder());
+ for (Integer index : sortCols.get(i).getIndexes()) {
+ // The only time this condition should be false is in the case of dynamic partitioning
+ if (index < colInfos.size()) {
+ sortCol.addAlias(colInfos.get(index).getInternalName(), index);
+ } else {
+ return null;
+ }
+ }
+ newSortCols.add(sortCol);
+ }
+ return newSortCols;
+ }
+
+ /**
+ * Processor for FileSink operator.
+ */
+ public static class FileSinkInferrer extends DefaultInferrer implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ BucketingSortingCtx bctx = (BucketingSortingCtx)procCtx;
+ FileSinkOperator fop = (FileSinkOperator)nd;
+
+ Operator<? extends OperatorDesc> parent = getParent(stack);
+ List<BucketCol> bucketCols = bctx.getBucketedCols(parent);
+ List<ColumnInfo> colInfos = fop.getSchema().getSignature();
+
+ // Set the inferred bucket columns for the file this FileSink produces
+ if (bucketCols != null) {
+ List<BucketCol> newBucketCols = getNewBucketCols(bucketCols, colInfos);
+ bctx.getBucketedColsByDirectory().put(fop.getConf().getDirName(), newBucketCols);
+ bctx.setBucketedCols(fop, newBucketCols);
+ }
+
+ List<SortCol> sortCols = bctx.getSortedCols(parent);
+
+ // Set the inferred sort columns for the file this FileSink produces
+ if (sortCols != null) {
+ List<SortCol> newSortCols = getNewSortCols(sortCols, colInfos);
+ bctx.getSortedColsByDirectory().put(fop.getConf().getDirName(), newSortCols);
+ bctx.setSortedCols(fop, newSortCols);
+ }
+
+ return null;
+ }
+
+ }
+
+ /**
+ * Processor for Extract operator.
+ *
+ * Only handles the case where the tree looks like
+ *
+ * ReduceSinkOperator --- ExtractOperator
+ *
+ * This is the case for distribute by, sort by, order by, cluster by operators.
+ */
+ public static class ExtractInferrer extends DefaultInferrer implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ BucketingSortingCtx bctx = (BucketingSortingCtx)procCtx;
+ ExtractOperator exop = (ExtractOperator)nd;
+
+ // As of writing this, there is no case where this could be false, this is just protection
+ // from possible future changes
+ if (exop.getParentOperators().size() != 1) {
+ return null;
+ }
+
+ Operator<? extends OperatorDesc> parent = exop.getParentOperators().get(0);
+
+ // The caller of this method should guarantee this
+ assert(parent instanceof ReduceSinkOperator);
+
+ ReduceSinkOperator rop = (ReduceSinkOperator)parent;
+
+ // Go through the set of partition columns, and find their representatives in the values
+ // These represent the bucketed columns
+ List<BucketCol> bucketCols = new ArrayList<BucketCol>();
+ for (int i = 0; i < rop.getConf().getPartitionCols().size(); i++) {
+ boolean valueColFound = false;
+ for (int j = 0; j < rop.getConf().getValueCols().size(); j++) {
+ if (new ExprNodeDescEqualityWrapper(rop.getConf().getValueCols().get(j)).equals(
+ new ExprNodeDescEqualityWrapper(rop.getConf().getPartitionCols().get(i)))) {
+
+ bucketCols.add(new BucketCol(
+ rop.getSchema().getSignature().get(j).getInternalName(), j));
+ valueColFound = true;
+ break;
+ }
+ }
+
+ // If the partition columns can't all be found in the values then the data is not bucketed
+ if (!valueColFound) {
+ bucketCols.clear();
+ break;
+ }
+ }
+
+ // Go through the set of key columns, and find their representatives in the values
+ // These represent the sorted columns
+ String sortOrder = rop.getConf().getOrder();
+ List<SortCol> sortCols = new ArrayList<SortCol>();
+ for (int i = 0; i < rop.getConf().getKeyCols().size(); i++) {
+ boolean valueColFound = false;
+ for (int j = 0; j < rop.getConf().getValueCols().size(); j++) {
+ if (new ExprNodeDescEqualityWrapper(rop.getConf().getValueCols().get(j)).equals(
+ new ExprNodeDescEqualityWrapper(rop.getConf().getKeyCols().get(i)))) {
+
+ sortCols.add(new SortCol(
+ rop.getSchema().getSignature().get(j).getInternalName(), j, sortOrder.charAt(i)));
+ valueColFound = true;
+ break;
+ }
+ }
+
+ // If the sorted columns can't all be found in the values then the data is only sorted on
+ // the columns seen up until now
+ if (!valueColFound) {
+ break;
+ }
+ }
+
+ List<ColumnInfo> colInfos = exop.getSchema().getSignature();
+
+ if (!bucketCols.isEmpty()) {
+ List<BucketCol> newBucketCols = getNewBucketCols(bucketCols, colInfos);
+ bctx.setBucketedCols(exop, newBucketCols);
+ }
+
+ if (!sortCols.isEmpty()) {
+ List<SortCol> newSortCols = getNewSortCols(sortCols, colInfos);
+ bctx.setSortedCols(exop, newSortCols);
+ }
+
+ return null;
+ }
+ }
+
+ /**
+ * Processor for GroupByOperator, the special case where it follows a ForwardOperator
+ *
+ * There is a multi group by optimization which puts multiple group by operators in a
+ * reducer when they share the same keys and are part of a multi insert query.
+ *
+ * In this case the tree should look like
+ * Group By Operator
+ * /
+ * ReduceSinkOperator - ForwardOperator --- ...
+ * \
+ * GroupByOperator
+ *
+ */
+
+ public static class MultiGroupByInferrer extends GroupByInferrer implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ BucketingSortingCtx bctx = (BucketingSortingCtx)procCtx;
+ GroupByOperator gop = (GroupByOperator)nd;
+
+ if (gop.getParentOperators().size() != 1) {
+ return null;
+ }
+
+ Operator<? extends OperatorDesc> fop = gop.getParentOperators().get(0);
+
+ // The caller of this method should guarantee this
+ assert(fop instanceof ForwardOperator);
+
+ if (fop.getParentOperators().size() != 1) {
+ return null;
+ }
+
+ Operator<? extends OperatorDesc> rop = fop.getParentOperators().get(0);
+
+ // The caller of this method should guarantee this
+ assert(rop instanceof ReduceSinkOperator);
+
+ processGroupByReduceSink((ReduceSinkOperator) rop, gop, bctx);
+
+ processForward(fop, bctx, rop);
+
+ return processGroupBy(fop, gop, bctx);
+ }
+ }
+
+ /**
+ * Processor for GroupBy operator.
+ *
+ * This handles the standard use of a group by operator, the tree should look like
+ *
+ * ReduceSinkOperator --- GroupByOperator
+ *
+ * It is up to the caller to guarantee the tree matches this pattern.
+ */
+ public static class GroupByInferrer extends DefaultInferrer implements NodeProcessor {
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ BucketingSortingCtx bctx = (BucketingSortingCtx)procCtx;
+ GroupByOperator gop = (GroupByOperator)nd;
+
+ // As of writing this, there is no case where this could be false, this is just protection
+ // from possible future changes
+ if (gop.getParentOperators().size() != 1) {
+ return null;
+ }
+
+ Operator<? extends OperatorDesc> rop = gop.getParentOperators().get(0);
+
+ // The caller of this method should guarantee this
+ assert(rop instanceof ReduceSinkOperator);
+
+ processGroupByReduceSink((ReduceSinkOperator) rop, gop, bctx);
+
+ return processGroupBy((ReduceSinkOperator)rop , gop, bctx);
+ }
+
+ /**
+ * Process the ReduceSinkOperator preceding a GroupByOperator to determine which columns
+ * are bucketed and sorted.
+ *
+ * @param rop
+ * @param gop
+ * @param bctx
+ */
+ protected void processGroupByReduceSink(ReduceSinkOperator rop, GroupByOperator gop,
+ BucketingSortingCtx bctx){
+
+ String sortOrder = rop.getConf().getOrder();
+ List<BucketCol> bucketCols = new ArrayList<BucketCol>();
+ List<SortCol> sortCols = new ArrayList<SortCol>();
+ assert rop.getConf().getKeyCols().size() <= rop.getSchema().getSignature().size();
+ // Group by operators select the key cols, so no need to find them in the values
+ for (int i = 0; i < rop.getConf().getKeyCols().size(); i++) {
+ String colName = rop.getSchema().getSignature().get(i).getInternalName();
+ bucketCols.add(new BucketCol(colName, i));
+ sortCols.add(new SortCol(colName, i, sortOrder.charAt(i)));
+ }
+ bctx.setBucketedCols(rop, bucketCols);
+ bctx.setSortedCols(rop, sortCols);
+ }
+
+ /**
+ * Process a GroupByOperator to determine which if any columns the output is bucketed and
+ * sorted by, assumes the columns output by the parent which are bucketed and sorted have
+ * already been determined.
+ *
+ * @param parent
+ * @param gop
+ * @param bctx
+ * @return
+ */
+ protected Object processGroupBy(Operator<? extends OperatorDesc> parent, GroupByOperator gop,
+ BucketingSortingCtx bctx) {
+ List<BucketCol> bucketCols = bctx.getBucketedCols(parent);
+ List<SortCol> sortCols = bctx.getSortedCols(parent);
+ List<ColumnInfo> colInfos = gop.getSchema().getSignature();
+
+ if (bucketCols == null) {
+ assert sortCols == null;
+ return null;
+ }
+
+ if (bucketCols.isEmpty()) {
+ assert sortCols.isEmpty();
+ return null;
+ }
+
+ BucketCol[] newBucketCols = new BucketCol[bucketCols.size()];
+ SortCol[] newSortCols = new SortCol[sortCols.size()];
+
+ findBucketingSortingColumns(gop.getConf().getKeys(), colInfos, bucketCols, sortCols,
+ newBucketCols, newSortCols);
+
+ setBucketingColsIfComplete(bctx, gop, newBucketCols);
+
+ setSortingColsIfComplete(bctx, gop, newSortCols);
+
+ return null;
+ }
+ }
+
+ /**
+ * Filter processor
+ */
+ public static class ForwardingInferrer extends DefaultInferrer implements NodeProcessor {
+ @SuppressWarnings("unchecked")
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+
+ processForward((Operator<? extends OperatorDesc>)nd, (BucketingSortingCtx)procCtx,
+ getParent(stack));
+
+ return null;
+ }
+ }
+
+ public static NodeProcessor getDefaultProc() {
+ return new DefaultInferrer();
+ }
+
+ public static NodeProcessor getJoinProc() {
+ return new JoinInferrer();
+ }
+
+ public static NodeProcessor getSelProc() {
+ return new SelectInferrer();
+ }
+
+ public static NodeProcessor getGroupByProc() {
+ return new GroupByInferrer();
+ }
+
+ public static NodeProcessor getFileSinkProc() {
+ return new FileSinkInferrer();
+ }
+
+ public static NodeProcessor getExtractProc() {
+ return new ExtractInferrer();
+ }
+
+ public static NodeProcessor getFilterProc() {
+ return new ForwardingInferrer();
+ }
+
+ public static NodeProcessor getLimitProc() {
+ return new ForwardingInferrer();
+ }
+
+ public static NodeProcessor getLateralViewForwardProc() {
+ return new ForwardingInferrer();
+ }
+
+ public static NodeProcessor getLateralViewJoinProc() {
+ return new ForwardingInferrer();
+ }
+
+ public static NodeProcessor getForwardProc() {
+ return new ForwardingInferrer();
+ }
+
+ public static NodeProcessor getMultiGroupByProc() {
+ return new MultiGroupByInferrer();
+ }
+}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/PhysicalOptimizer.java Wed Jan 30 04:20:39 2013
@@ -59,6 +59,13 @@ public class PhysicalOptimizer {
if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVEMETADATAONLYQUERIES)) {
resolvers.add(new MetadataOnlyOptimizer());
}
+
+ // Physical optimizers which follow this need to be careful not to invalidate the inferences
+ // made by this optimizer. Only optimizers which depend on the results of this one should
+ // follow it.
+ if (hiveConf.getBoolVar(HiveConf.ConfVars.HIVE_INFER_BUCKET_SORT)) {
+ resolvers.add(new BucketingSortingInferenceOptimizer());
+ }
}
/**
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredWork.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredWork.java?rev=1440271&r1=1440270&r2=1440271&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredWork.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/MapredWork.java Wed Jan 30 04:20:39 2013
@@ -31,6 +31,8 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.Operator;
import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.BucketCol;
+import org.apache.hadoop.hive.ql.optimizer.physical.BucketingSortingCtx.SortCol;
import org.apache.hadoop.hive.ql.parse.OpParseContext;
import org.apache.hadoop.hive.ql.parse.QBJoinTree;
import org.apache.hadoop.hive.ql.parse.SplitSample;
@@ -93,6 +95,18 @@ public class MapredWork extends Abstract
private transient boolean useBucketizedHiveInputFormat;
+ // if this is true, this means that this is the map reduce task which writes the final data,
+ // ignoring the optional merge task
+ private boolean finalMapRed = false;
+
+ // If this map reduce task has a FileSinkOperator, and bucketing/sorting metadata can be
+ // inferred about the data being written by that operator, these are mappings from the directory
+ // that operator writes into to the bucket/sort columns for that data.
+ private final Map<String, List<BucketCol>> bucketedColsByDirectory =
+ new HashMap<String, List<BucketCol>>();
+ private final Map<String, List<SortCol>> sortedColsByDirectory =
+ new HashMap<String, List<SortCol>>();
+
public MapredWork() {
aliasToPartnInfo = new LinkedHashMap<String, PartitionDesc>();
}
@@ -279,6 +293,16 @@ public class MapredWork extends Abstract
this.numReduceTasks = numReduceTasks;
}
+ @Explain(displayName = "Path -> Bucketed Columns", normalExplain = false)
+ public Map<String, List<BucketCol>> getBucketedColsByDirectory() {
+ return bucketedColsByDirectory;
+ }
+
+ @Explain(displayName = "Path -> Sorted Columns", normalExplain = false)
+ public Map<String, List<SortCol>> getSortedColsByDirectory() {
+ return sortedColsByDirectory;
+ }
+
@SuppressWarnings("nls")
public void addMapWork(String path, String alias, Operator<?> work,
PartitionDesc pd) {
@@ -525,4 +549,12 @@ public class MapredWork extends Abstract
public void setUseBucketizedHiveInputFormat(boolean useBucketizedHiveInputFormat) {
this.useBucketizedHiveInputFormat = useBucketizedHiveInputFormat;
}
+
+ public boolean isFinalMapRed() {
+ return finalMapRed;
+ }
+
+ public void setFinalMapRed(boolean finalMapRed) {
+ this.finalMapRed = finalMapRed;
+ }
}
Added: hive/trunk/ql/src/test/queries/clientnegative/merge_negative_3.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientnegative/merge_negative_3.q?rev=1440271&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientnegative/merge_negative_3.q (added)
+++ hive/trunk/ql/src/test/queries/clientnegative/merge_negative_3.q Wed Jan 30 04:20:39 2013
@@ -0,0 +1,6 @@
+set hive.enforce.bucketing=true;
+set hive.enforce.sorting=true;
+
+create table srcpart2 (key int, value string) partitioned by (ds string) clustered by (key) sorted by (key) into 2 buckets stored as RCFILE;
+insert overwrite table srcpart2 partition (ds='2011') select * from src;
+alter table srcpart2 partition (ds = '2011') concatenate;