You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ke...@apache.org on 2012/10/31 00:39:19 UTC
svn commit: r1403928 [1/7] - in /hive/trunk:
common/src/java/org/apache/hadoop/hive/conf/ conf/
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/metadata/ ql/src/java/org/apache/ha...
Author: kevinwilfong
Date: Tue Oct 30 23:39:17 2012
New Revision: 1403928
URL: http://svn.apache.org/viewvc?rev=1403928&view=rev
Log:
HIVE-3276. optimize union sub-queries. (njain via kevinwilfong)
Added:
hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_1.q
hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_2.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_1.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_10.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_11.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_12.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_13.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_14.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_15.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_16.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_17.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_18.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_19.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_2.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_20.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_21.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_22.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_23.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_24.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_3.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_4.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_5.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_6.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_7.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_8.q
hive/trunk/ql/src/test/queries/clientpositive/union_remove_9.q
hive/trunk/ql/src/test/results/clientpositive/skewjoin_union_remove_1.q.out
hive/trunk/ql/src/test/results/clientpositive/skewjoin_union_remove_2.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_1.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_10.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_11.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_12.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_13.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_14.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_15.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_16.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_17.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_18.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_19.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_2.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_20.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_21.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_22.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_23.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_24.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_3.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_4.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_5.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_6.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_7.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_8.q.out
hive/trunk/ql/src/test/results/clientpositive/union_remove_9.q.out
Modified:
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/ErrorMsg.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FilterOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SelectOperator.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Table.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/GenMRProcContext.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcContext.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcessor.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java
hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/SelectDesc.java
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=1403928&r1=1403927&r2=1403928&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 Tue Oct 30 23:39:17 2012
@@ -510,6 +510,15 @@ public class HiveConf extends Configurat
HIVEOPTBUCKETMAPJOIN("hive.optimize.bucketmapjoin", false), // optimize bucket map join
HIVEOPTSORTMERGEBUCKETMAPJOIN("hive.optimize.bucketmapjoin.sortedmerge", false), // try to use sorted merge bucket map join
HIVEOPTREDUCEDEDUPLICATION("hive.optimize.reducededuplication", true),
+ // whether to optimize union followed by select followed by filesink
+ // It creates sub-directories in the final output, so should not be turned on in systems
+ // where MAPREDUCE-1501 is not present
+ HIVE_OPTIMIZE_UNION_REMOVE("hive.optimize.union.remove", false),
+
+ // whether hadoop map-reduce supports sub-directories. It was added by MAPREDUCE-1501.
+ // Some optimizations can only be performed if the version of hadoop being used supports
+ // sub-directories
+ HIVE_HADOOP_SUPPORTS_SUBDIRECTORIES("hive.mapred.supports.subdirectories", false),
// optimize skewed join by changing the query plan at compile time
HIVE_OPTIMIZE_SKEWJOIN_COMPILETIME("hive.optimize.skewjoin.compiletime", false),
Modified: hive/trunk/conf/hive-default.xml.template
URL: http://svn.apache.org/viewvc/hive/trunk/conf/hive-default.xml.template?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/conf/hive-default.xml.template (original)
+++ hive/trunk/conf/hive-default.xml.template Tue Oct 30 23:39:17 2012
@@ -453,6 +453,29 @@
</property>
<property>
+ <name>hive.optimize.union.remove</name>
+ <value>false</value>
+ <description>
+ Whether to remove the union and push the operators between union and the filesink above
+ union. This avoids an extra scan of the output by union. This is independently useful for union
+ queries, and specially useful when hive.optimize.skewjoin.compiletime is set to true, since an
+ extra union is inserted.
+
+ The merge is triggered if either of hive.merge.mapfiles or hive.merge.mapredfiles is set to true.
+ If the user has set hive.merge.mapfiles to true and hive.merge.mapredfiles to false, the idea was the
+ number of reducers are few, so the number of files anyway are small. However, with this optimization,
+ we are increasing the number of files possibly by a big margin. So, we merge aggresively.
+</property>
+
+<property>
+ <name>hive.mapred.supports.subdirectories</name>
+ <value>false</value>
+ <description>Whether the version of hadoop which is running supports sub-directories for tables/partitions.
+ Many hive optimizations can be applied if the hadoop version supports sub-directories for
+ tables/partitions. It was added by MAPREDUCE-1501
+</property>
+
+<property>
<name>hive.multigroupby.singlemr</name>
<value>false</value>
<description>Whether to optimize multi group by query to generate single M/R
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java Tue Oct 30 23:39:17 2012
@@ -270,6 +270,12 @@ public enum ErrorMsg {
"Grouping sets aggregations (with rollups or cubes) are not allowed if aggregation function " +
"parameters overlap with the aggregation functions columns"),
+ HIVE_UNION_REMOVE_OPTIMIZATION_NEEDS_SUBDIRECTORIES(10212,
+ "In order to use hive.optimize.union.remove, the hadoop version that you are using " +
+ "should support sub-directories for tables/partitions. If that is true, set " +
+ "hive.hadoop.supports.subdirectories to true. Otherwise, set hive.optimize.union.remove " +
+ "to false"),
+
SCRIPT_INIT_ERROR(20000, "Unable to initialize custom script."),
SCRIPT_IO_ERROR(20001, "An error occurred while reading or writing to your custom script. "
+ "It may have crashed with an error."),
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/ColumnInfo.java Tue Oct 30 23:39:17 2012
@@ -180,4 +180,27 @@ public class ColumnInfo implements Seria
public void setSkewedCol(boolean isSkewedCol) {
this.isSkewedCol = isSkewedCol;
}
+
+ private boolean checkEquals(Object obj1, Object obj2) {
+ return obj1 == null ? obj2 == null : obj1.equals(obj2);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (!(obj instanceof ColumnInfo) || (obj == null)) {
+ return false;
+ }
+
+ ColumnInfo dest = (ColumnInfo)obj;
+ if ((!checkEquals(internalName, dest.getInternalName())) ||
+ (!checkEquals(alias, dest.getAlias())) ||
+ (!checkEquals(getType(), dest.getType())) ||
+ (isSkewedCol != dest.isSkewedCol()) ||
+ (isVirtualCol != dest.getIsVirtualCol()) ||
+ (isHiddenVirtualCol != dest.isHiddenVirtualCol())) {
+ return false;
+ }
+
+ return true;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java Tue Oct 30 23:39:17 2012
@@ -77,6 +77,7 @@ public class FileSinkOperator extends Te
protected transient Path parent;
protected transient HiveOutputFormat<?, ?> hiveOutputFormat;
protected transient Path specPath;
+ protected transient String childSpecPathDynLinkedPartitions;
protected transient int dpStartCol; // start column # for DP columns
protected transient List<String> dpVals; // array of values corresponding to DP columns
protected transient List<Object> dpWritables;
@@ -119,16 +120,6 @@ public class FileSinkOperator extends Te
}
/**
- * Append a subdirectory to the tmp path.
- *
- * @param dp
- * subdirecgtory name
- */
- public void appendTmpPath(String dp) {
- tmpPath = new Path(tmpPath, dp);
- }
-
- /**
* Update OutPath according to tmpPath.
*/
public Path getTaskOutPath(String taskId) {
@@ -282,6 +273,30 @@ public class FileSinkOperator extends Te
private boolean filesCreated = false;
+ private void initializeSpecPath() {
+ // For a query of the type:
+ // insert overwrite table T1
+ // select * from (subq1 union all subq2)u;
+ // subQ1 and subQ2 write to directories Parent/Child_1 and
+ // Parent/Child_2 respectively, and union is removed.
+ // The movetask that follows subQ1 and subQ2 tasks moves the directory
+ // 'Parent'
+
+ // However, if the above query contains dynamic partitions, subQ1 and
+ // subQ2 have to write to directories: Parent/DynamicPartition/Child_1
+ // and Parent/DynamicPartition/Child_1 respectively.
+ // The movetask that follows subQ1 and subQ2 tasks still moves the directory
+ // 'Parent'
+ if ((!conf.isLinkedFileSink()) || (dpCtx == null)) {
+ specPath = new Path(conf.getDirName());
+ childSpecPathDynLinkedPartitions = null;
+ return;
+ }
+
+ specPath = new Path(conf.getParentDir());
+ childSpecPathDynLinkedPartitions = Utilities.getFileNameFromDirName(conf.getDirName());
+ }
+
@Override
protected void initializeOp(Configuration hconf) throws HiveException {
try {
@@ -294,7 +309,7 @@ public class FileSinkOperator extends Te
dpCtx = conf.getDynPartCtx();
valToPaths = new HashMap<String, FSPaths>();
taskId = Utilities.getTaskId(hconf);
- specPath = new Path(conf.getDirName());
+ initializeSpecPath();
fs = specPath.getFileSystem(hconf);
hiveOutputFormat = conf.getTableInfo().getOutputFileFormatClass().newInstance();
isCompressed = conf.getCompressed();
@@ -625,8 +640,17 @@ public class FileSinkOperator extends Te
LOG.error("Fatal error was thrown due to exceeding number of dynamic partitions");
}
fsp2 = new FSPaths(specPath);
- fsp2.tmpPath = new Path(fsp2.tmpPath, dpDir);
- fsp2.taskOutputTempPath = new Path(fsp2.taskOutputTempPath, dpDir);
+ if (childSpecPathDynLinkedPartitions != null) {
+ fsp2.tmpPath = new Path(fsp2.tmpPath,
+ dpDir + Path.SEPARATOR + childSpecPathDynLinkedPartitions);
+ fsp2.taskOutputTempPath =
+ new Path(fsp2.taskOutputTempPath,
+ dpDir + Path.SEPARATOR + childSpecPathDynLinkedPartitions);
+ } else {
+ fsp2.tmpPath = new Path(fsp2.tmpPath, dpDir);
+ fsp2.taskOutputTempPath =
+ new Path(fsp2.taskOutputTempPath, dpDir);
+ }
createBucketFiles(fsp2);
valToPaths.put(dpDir, fsp2);
}
@@ -668,7 +692,11 @@ public class FileSinkOperator extends Te
DynamicPartitionCtx dpCtx = conf.getDynPartCtx();
int numDP = dpCtx.getNumDPCols();
FileSystem fs = tmpPath.getFileSystem(jobConf);
- FileStatus[] status = Utilities.getFileStatusRecurse(tmpPath, numDP, fs);
+ int level = numDP;
+ if (conf.isLinkedFileSink()) {
+ level++;
+ }
+ FileStatus[] status = Utilities.getFileStatusRecurse(tmpPath, level, fs);
sb.append("Sample of ")
.append(Math.min(status.length, 100))
.append(" partitions created under ")
@@ -750,6 +778,9 @@ public class FileSinkOperator extends Te
if ((conf != null) && isNativeTable) {
String specPath = conf.getDirName();
DynamicPartitionCtx dpCtx = conf.getDynPartCtx();
+ if (conf.isLinkedFileSink() && (dpCtx != null)) {
+ specPath = conf.getParentDir();
+ }
Utilities.mvFileToFinalPath(specPath, hconf, success, LOG, dpCtx, conf);
}
} catch (IOException e) {
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FilterOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FilterOperator.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FilterOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/FilterOperator.java Tue Oct 30 23:39:17 2012
@@ -170,4 +170,9 @@ public class FilterOperator extends Oper
public boolean columnNamesRowResolvedCanBeObtained() {
return true;
}
+
+ @Override
+ public boolean supportUnionRemoveOptimization() {
+ return true;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Operator.java Tue Oct 30 23:39:17 2012
@@ -1420,4 +1420,8 @@ public abstract class Operator<T extends
public void setUseBucketizedHiveInputFormat(boolean useBucketizedHiveInputFormat) {
this.useBucketizedHiveInputFormat = useBucketizedHiveInputFormat;
}
+
+ public boolean supportUnionRemoveOptimization() {
+ return false;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SelectOperator.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SelectOperator.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SelectOperator.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/SelectOperator.java Tue Oct 30 23:39:17 2012
@@ -110,4 +110,9 @@ public class SelectOperator extends Oper
public boolean columnNamesRowResolvedCanBeObtained() {
return true;
}
+
+ @Override
+ public boolean supportUnionRemoveOptimization() {
+ return true;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java Tue Oct 30 23:39:17 2012
@@ -1135,18 +1135,25 @@ public final class Utilities {
// move file by file
FileStatus[] files = fs.listStatus(src);
for (FileStatus file : files) {
+
Path srcFilePath = file.getPath();
String fileName = srcFilePath.getName();
Path dstFilePath = new Path(dst, fileName);
- if (fs.exists(dstFilePath)) {
- int suffix = 0;
- do {
- suffix++;
- dstFilePath = new Path(dst, fileName + "_" + suffix);
- } while (fs.exists(dstFilePath));
+ if (file.isDir()) {
+ renameOrMoveFiles(fs, srcFilePath, dstFilePath);
}
- if (!fs.rename(srcFilePath, dstFilePath)) {
- throw new HiveException("Unable to move: " + src + " to: " + dst);
+ else {
+ if (fs.exists(dstFilePath)) {
+ int suffix = 0;
+ do {
+ suffix++;
+ dstFilePath = new Path(dst, fileName + "_" + suffix);
+ } while (fs.exists(dstFilePath));
+ }
+
+ if (!fs.rename(srcFilePath, dstFilePath)) {
+ throw new HiveException("Unable to move: " + src + " to: " + dst);
+ }
}
}
}
@@ -1213,6 +1220,14 @@ public final class Utilities {
return taskId;
}
+ public static String getFileNameFromDirName(String dirName) {
+ int dirEnd = dirName.lastIndexOf(Path.SEPARATOR);
+ if (dirEnd != -1) {
+ return dirName.substring(dirEnd + 1);
+ }
+ return dirName;
+ }
+
/**
* Replace the task id from the filename. It is assumed that the filename is derived from the
* output of getTaskId
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=1403928&r1=1403927&r2=1403928&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 Tue Oct 30 23:39:17 2012
@@ -1180,7 +1180,7 @@ public class Hive {
Hive.replaceFiles(loadPath, newPartPath, oldPartPath, getConf());
} else {
FileSystem fs = FileSystem.get(tbl.getDataLocation(), getConf());
- Hive.copyFiles(loadPath, newPartPath, fs);
+ Hive.copyFiles(conf, loadPath, newPartPath, fs);
}
// recreate the partition if it existed before
@@ -1224,7 +1224,9 @@ public class Hive {
FileStatus[] leafStatus = Utilities.getFileStatusRecurse(loadPath, numDP+1, fs);
// Check for empty partitions
for (FileStatus s : leafStatus) {
- if (s.isDir()) {
+ // Check if the hadoop version supports sub-directories for tables/partitions
+ if (s.isDir() &&
+ !conf.getBoolVar(HiveConf.ConfVars.HIVE_HADOOP_SUPPORTS_SUBDIRECTORIES)) {
// No leaves in this directory
LOG.info("NOT moving empty directory: " + s.getPath());
} else {
@@ -1888,8 +1890,9 @@ public class Hive {
}
}
- static private void checkPaths(FileSystem fs, FileStatus[] srcs, Path destf,
- boolean replace) throws HiveException {
+ static private void checkPaths(HiveConf conf,
+ FileSystem fs, FileStatus[] srcs, Path destf,
+ boolean replace) throws HiveException {
try {
for (FileStatus src : srcs) {
FileStatus[] items = fs.listStatus(src.getPath());
@@ -1903,7 +1906,9 @@ public class Hive {
fs.delete(itemStaging, true);
continue;
}
- if (item.isDir()) {
+
+ if (!conf.getBoolVar(HiveConf.ConfVars.HIVE_HADOOP_SUPPORTS_SUBDIRECTORIES) &&
+ item.isDir()) {
throw new HiveException("checkPaths: " + src.getPath()
+ " has nested directory" + itemStaging);
}
@@ -1963,7 +1968,7 @@ public class Hive {
}
}
- static protected void copyFiles(Path srcf, Path destf, FileSystem fs)
+ static protected void copyFiles(HiveConf conf, Path srcf, Path destf, FileSystem fs)
throws HiveException {
try {
// create the destination if it does not exist
@@ -1989,7 +1994,7 @@ public class Hive {
// srcs = new FileStatus[0]; Why is this needed?
}
// check that source and target paths exist
- checkPaths(fs, srcs, destf, false);
+ checkPaths(conf, fs, srcs, destf, false);
// move it, move it
try {
@@ -2024,7 +2029,7 @@ public class Hive {
* The directory where the old data location, need to be cleaned up.
*/
static protected void replaceFiles(Path srcf, Path destf, Path oldPath,
- Configuration conf) throws HiveException {
+ HiveConf conf) throws HiveException {
try {
FileSystem fs = srcf.getFileSystem(conf);
@@ -2040,7 +2045,7 @@ public class Hive {
LOG.info("No sources specified to move: " + srcf);
return;
}
- checkPaths(fs, srcs, destf, true);
+ checkPaths(conf, fs, srcs, destf, true);
// point of no return -- delete oldPath
if (oldPath != null) {
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Table.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Table.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Table.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/metadata/Table.java Tue Oct 30 23:39:17 2012
@@ -622,7 +622,7 @@ public class Table implements Serializab
FileSystem fs;
try {
fs = FileSystem.get(getDataLocation(), Hive.get().getConf());
- Hive.copyFiles(srcf, new Path(getDataLocation().getPath()), fs);
+ Hive.copyFiles(Hive.get().getConf(), srcf, new Path(getDataLocation().getPath()), fs);
} catch (IOException e) {
throw new HiveException("addFiles: filesystem error in check phase", e);
}
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=1403928&r1=1403927&r2=1403928&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 Tue Oct 30 23:39:17 2012
@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Stack;
import org.apache.commons.logging.Log;
@@ -109,6 +110,14 @@ public class GenMRFileSink1 implements N
parseCtx.getQB().getParseInfo().isInsertToTable();
HiveConf hconf = parseCtx.getConf();
+ // 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();
+ if (fileSinkDescs != null) {
+ Task<? extends Serializable> childTask = fileSinkDescs.get(fsOp.getConf());
+ processLinkedFileDesc(ctx, childTask);
+ return null;
+ }
// Has the user enabled merging of files for map-only jobs or for all jobs
if ((ctx.getMvTask() != null) && (!ctx.getMvTask().isEmpty())) {
@@ -129,18 +138,30 @@ public class GenMRFileSink1 implements N
}
if ((mvTask != null) && !mvTask.isLocal()) {
- // There are separate configuration parameters to control whether to
- // merge for a map-only job
- // or for a map-reduce job
- MapredWork currWork = (MapredWork) currTask.getWork();
- boolean mergeMapOnly =
- hconf.getBoolVar(HiveConf.ConfVars.HIVEMERGEMAPFILES) &&
- currWork.getReducer() == null;
- boolean mergeMapRed =
- hconf.getBoolVar(HiveConf.ConfVars.HIVEMERGEMAPREDFILES) &&
- currWork.getReducer() != null;
- if (mergeMapOnly || mergeMapRed) {
- chDir = true;
+ if (fsOp.getConf().isLinkedFileSink()) {
+ // If the user has HIVEMERGEMAPREDFILES set to false, the idea was the
+ // number of reducers are few, so the number of files anyway are small.
+ // However, with this optimization, we are increasing the number of files
+ // possibly by a big margin. So, merge aggresively.
+ if (hconf.getBoolVar(HiveConf.ConfVars.HIVEMERGEMAPFILES) ||
+ hconf.getBoolVar(HiveConf.ConfVars.HIVEMERGEMAPREDFILES)) {
+ chDir = true;
+ }
+ }
+ else {
+ // There are separate configuration parameters to control whether to
+ // merge for a map-only job
+ // or for a map-reduce job
+ MapredWork currWork = (MapredWork) currTask.getWork();
+ boolean mergeMapOnly =
+ hconf.getBoolVar(HiveConf.ConfVars.HIVEMERGEMAPFILES) &&
+ currWork.getReducer() == null;
+ boolean mergeMapRed =
+ hconf.getBoolVar(HiveConf.ConfVars.HIVEMERGEMAPREDFILES) &&
+ currWork.getReducer() != null;
+ if (mergeMapOnly || mergeMapRed) {
+ chDir = true;
+ }
}
}
}
@@ -153,9 +174,57 @@ public class GenMRFileSink1 implements N
createMergeJob((FileSinkOperator) nd, ctx, finalName);
}
+ FileSinkDesc fileSinkDesc = fsOp.getConf();
+ if (fileSinkDesc.isLinkedFileSink()) {
+ Map<FileSinkDesc, Task<? extends Serializable>> linkedFileDescTasks =
+ ctx.getLinkedFileDescTasks();
+ if (linkedFileDescTasks == null) {
+ linkedFileDescTasks = new HashMap<FileSinkDesc, Task<? extends Serializable>>();
+ ctx.setLinkedFileDescTasks(linkedFileDescTasks);
+ }
+
+ // The child tasks may be null in case of a select
+ if ((currTask.getChildTasks() != null) &&
+ (currTask.getChildTasks().size() == 1)) {
+ for (FileSinkDesc fileDesc : fileSinkDesc.getLinkedFileSinkDesc()) {
+ linkedFileDescTasks.put(fileDesc, currTask.getChildTasks().get(0));
+ }
+ }
+ }
+
return null;
}
+ /*
+ * Multiple file sink descriptors are linked.
+ * Use the task created by the first linked file descriptor
+ */
+ private void processLinkedFileDesc(GenMRProcContext ctx,
+ Task<? extends Serializable> childTask)
+ throws SemanticException {
+ Operator<? extends OperatorDesc> currTopOp = ctx.getCurrTopOp();
+ String currAliasId = ctx.getCurrAliasId();
+ List<Operator<? extends OperatorDesc>> seenOps = ctx.getSeenOps();
+ List<Task<? extends Serializable>> rootTasks = ctx.getRootTasks();
+ Task<? extends Serializable> currTask = ctx.getCurrTask();
+
+ if (currTopOp != null) {
+ if (!seenOps.contains(currTopOp)) {
+ seenOps.add(currTopOp);
+ GenMapRedUtils.setTaskPlan(currAliasId, currTopOp,
+ (MapredWork) currTask.getWork(), false, ctx);
+ }
+
+ if (!rootTasks.contains(currTask)) {
+ rootTasks.add(currTask);
+ }
+ }
+
+ if (childTask != null) {
+ currTask.addDependentTask(childTask);
+ }
+ }
+
/**
* Add the StatsTask as a dependent task of the MoveTask
* because StatsTask will change the Table/Partition metadata. For atomicity, we
@@ -262,10 +331,10 @@ public class GenMRFileSink1 implements N
// since it is unknown if the merge MR will be triggered at execution time.
MoveWork dummyMv = new MoveWork(null, null, null,
- new LoadFileDesc(fsConf.getDirName(), finalName, true, null, null), false);
+ new LoadFileDesc(fsConf.getFinalDirName(), finalName, true, null, null), false);
ConditionalTask cndTsk = createCondTask(conf, currTask, dummyMv, cplan,
- fsConf.getDirName());
+ fsConf.getFinalDirName());
linkMoveTask(ctx, newOutput, cndTsk);
}
@@ -387,7 +456,7 @@ public class GenMRFileSink1 implements N
//
MapRedTask currTask = (MapRedTask) ctx.getCurrTask();
MoveWork dummyMv = new MoveWork(null, null, null,
- new LoadFileDesc(fsInputDesc.getDirName(), finalName, true, null, null), false);
+ new LoadFileDesc(fsInputDesc.getFinalDirName(), finalName, true, null, null), false);
MapredWork cplan;
if(parseCtx.getConf().getBoolVar(HiveConf.ConfVars.
@@ -417,7 +486,7 @@ public class GenMRFileSink1 implements N
// NOTE: we should gather stats in MR1 rather than MR2 at merge job since we don't
// know if merge MR2 will be triggered at execution time
ConditionalTask cndTsk = createCondTask(ctx.getConf(), ctx.getCurrTask(), dummyMv, cplan,
- fsInputDesc.getDirName());
+ fsInputDesc.getFinalDirName());
// keep the dynamic partition context in conditional task resolver context
ConditionalResolverMergeFilesCtx mrCtx =
@@ -516,7 +585,7 @@ public class GenMRFileSink1 implements N
FileSinkDesc fsDesc) {
ArrayList<String> aliases = new ArrayList<String>();
- String inputDir = fsDesc.getDirName();
+ String inputDir = fsDesc.getFinalDirName();
TableDesc tblDesc = fsDesc.getTableInfo();
aliases.add(inputDir); // dummy alias: just use the input path
@@ -541,7 +610,7 @@ public class GenMRFileSink1 implements N
private MapredWork createRCFileMergeTask(FileSinkDesc fsInputDesc,
String finalName, boolean hasDynamicPartitions) throws SemanticException {
- String inputDir = fsInputDesc.getDirName();
+ String inputDir = fsInputDesc.getFinalDirName();
TableDesc tblDesc = fsInputDesc.getTableInfo();
if(tblDesc.getInputFileFormatClass().equals(RCFileInputFormat.class)) {
@@ -637,8 +706,9 @@ public class GenMRFileSink1 implements N
srcDir = mvWork.getLoadTableWork().getSourceDir();
}
+ String fsOpDirName = fsOp.getConf().getFinalDirName();
if ((srcDir != null)
- && (srcDir.equalsIgnoreCase(fsOp.getConf().getDirName()))) {
+ && (srcDir.equalsIgnoreCase(fsOpDirName))) {
return mvTsk;
}
}
@@ -681,7 +751,7 @@ public class GenMRFileSink1 implements N
String dest = null;
if (chDir) {
- dest = fsOp.getConf().getDirName();
+ dest = fsOp.getConf().getFinalDirName();
// generate the temporary file
// it must be on the same file system as the current destination
@@ -689,7 +759,17 @@ public class GenMRFileSink1 implements N
Context baseCtx = parseCtx.getContext();
String tmpDir = baseCtx.getExternalTmpFileURI((new Path(dest)).toUri());
- fsOp.getConf().setDirName(tmpDir);
+ FileSinkDesc fileSinkDesc = fsOp.getConf();
+ // Change all the linked file sink descriptors
+ if (fileSinkDesc.isLinkedFileSink()) {
+ for (FileSinkDesc fsConf:fileSinkDesc.getLinkedFileSinkDesc()) {
+ String fileName = Utilities.getFileNameFromDirName(fsConf.getDirName());
+ fsConf.setParentDir(tmpDir);
+ fsConf.setDirName(tmpDir + Path.SEPARATOR + fileName);
+ }
+ } else {
+ fileSinkDesc.setDirName(tmpDir);
+ }
}
Task<MoveWork> mvTask = null;
@@ -707,7 +787,6 @@ public class GenMRFileSink1 implements N
// Set the move task to be dependent on the current task
if (mvTask != null) {
-
addDependentMoveTasks(ctx, mvTask, currTask);
}
@@ -742,7 +821,7 @@ public class GenMRFileSink1 implements N
}
}
// mapTask and currTask should be merged by and join/union operator
- // (e.g., GenMRUnion1j) which has multiple topOps.
+ // (e.g., GenMRUnion1) which has multiple topOps.
// assert mapTask == currTask : "mapTask.id = " + mapTask.getId()
// + "; currTask.id = " + currTask.getId();
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRProcContext.java Tue Oct 30 23:39:17 2012
@@ -23,6 +23,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -38,6 +39,7 @@ import org.apache.hadoop.hive.ql.hooks.W
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
import org.apache.hadoop.hive.ql.parse.ParseContext;
import org.apache.hadoop.hive.ql.plan.DependencyCollectionWork;
+import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.MapJoinDesc;
import org.apache.hadoop.hive.ql.plan.MoveWork;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
@@ -253,6 +255,10 @@ public class GenMRProcContext implements
private List<Operator<? extends OperatorDesc>> rootOps;
private DependencyCollectionTask dependencyTaskForMultiInsert;
+ // If many fileSinkDescs are linked to each other, it is a good idea to keep track of
+ // tasks for first fileSinkDesc. others can use it
+ private Map<FileSinkDesc, Task<? extends Serializable>> linkedFileDescTasks;
+
/**
* Set of read entities. This list is generated by the walker and is passed to
* the hooks.
@@ -314,6 +320,7 @@ public class GenMRProcContext implements
unionTaskMap = new HashMap<UnionOperator, GenMRUnionCtx>();
mapJoinTaskMap = new HashMap<AbstractMapJoinOperator<? extends MapJoinDesc>, GenMRMapJoinCtx>();
dependencyTaskForMultiInsert = null;
+ linkedFileDescTasks = null;
}
/**
@@ -568,4 +575,13 @@ public class GenMRProcContext implements
}
return dependencyTaskForMultiInsert;
}
+
+ public Map<FileSinkDesc, Task<? extends Serializable>> getLinkedFileDescTasks() {
+ return linkedFileDescTasks;
+ }
+
+ public void setLinkedFileDescTasks(
+ Map<FileSinkDesc, Task<? extends Serializable>> linkedFileDescTasks) {
+ this.linkedFileDescTasks = linkedFileDescTasks;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcContext.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcContext.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcContext.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcContext.java Tue Oct 30 23:39:17 2012
@@ -23,6 +23,7 @@ import java.util.Map;
import org.apache.hadoop.hive.ql.exec.UnionOperator;
import org.apache.hadoop.hive.ql.lib.NodeProcessorCtx;
+import org.apache.hadoop.hive.ql.parse.ParseContext;
/**
* UnionProcContext.
@@ -112,12 +113,14 @@ public class UnionProcContext implements
}
return true;
}
-
}
// the subqueries are map-only jobs
private boolean mapOnlySubq;
+ // ParseContext
+ private ParseContext parseContext;
+
/**
* @return the mapOnlySubq
*/
@@ -147,4 +150,12 @@ public class UnionProcContext implements
public UnionParseContext getUnionParseContext(UnionOperator u) {
return uCtxMap.get(u);
}
+
+ public ParseContext getParseContext() {
+ return parseContext;
+ }
+
+ public void setParseContext(ParseContext parseContext) {
+ this.parseContext = parseContext;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java Tue Oct 30 23:39:17 2012
@@ -17,16 +17,21 @@
*/
package org.apache.hadoop.hive.ql.optimizer.unionproc;
+import java.util.ArrayList;
import java.util.List;
import java.util.Stack;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.ql.exec.OperatorFactory;
import org.apache.hadoop.hive.ql.exec.UnionOperator;
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.optimizer.unionproc.UnionProcContext.UnionParseContext;
import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.hive.ql.plan.FileSinkDesc;
import org.apache.hadoop.hive.ql.plan.OperatorDesc;
/**
@@ -146,7 +151,7 @@ public final class UnionProcFactory {
UnionOperator parentUnionOperator = null;
while (start >= 0) {
Operator<? extends OperatorDesc> parent =
- (Operator<? extends OperatorDesc>) stack.get(start);
+ (Operator<? extends OperatorDesc>) stack.get(start);
if (parent instanceof UnionOperator) {
parentUnionOperator = (UnionOperator) parent;
break;
@@ -158,7 +163,7 @@ public final class UnionProcFactory {
boolean mapOnly = false;
if (parentUnionOperator != null) {
UnionParseContext parentUCtx =
- ctx.getUnionParseContext(parentUnionOperator);
+ ctx.getUnionParseContext(parentUnionOperator);
if (parentUCtx != null && parentUCtx.allMapOnlySubQSet()) {
mapOnly = parentUCtx.allMapOnlySubQ();
}
@@ -173,6 +178,139 @@ public final class UnionProcFactory {
}
/**
+ * Union followed by no processing.
+ * This is to optimize queries of the type:
+ * select * from (subq1 union all subq2 ...)x;
+ * where at least one of the queries involve a map-reduce job.
+ * There is no need for a union in this scenario - it involves an extra
+ * write and read for the final output without this optimization.
+ * Queries of the form:
+ * select x.c1 from (subq1 union all subq2 ...)x where filter(x.c2);
+ * can be transformed to:
+ * select * from (subq1 where filter union all subq2 where filter ...)x;
+ * and then optimized.
+ */
+ public static class UnionNoProcessFile implements NodeProcessor {
+
+ private void pushOperatorsAboveUnion(UnionOperator union,
+ Stack<Node> stack, int pos) throws SemanticException {
+ // Clone all the operators between union and filescan, and push them above
+ // the union. Remove the union (the tree below union gets delinked after that)
+ try {
+ List<Operator<? extends OperatorDesc>> parents =
+ union.getParentOperators();
+ int numParents = parents.size();
+ for (Operator<? extends OperatorDesc> parent : parents) {
+ parent.setChildOperators(null);
+ }
+
+ for (; pos < stack.size() - 1; pos++) {
+ Operator<? extends OperatorDesc> originalOp =
+ (Operator<? extends OperatorDesc>)stack.get(pos);
+
+ for (int p = 0; p < numParents; p++) {
+ OperatorDesc cloneDesc = (OperatorDesc)originalOp.getConf().clone();
+
+ Operator<? extends OperatorDesc> cloneOp =
+ OperatorFactory.getAndMakeChild(cloneDesc, originalOp.getSchema(), parents.get(p));
+ parents.set(p, cloneOp);
+ }
+ }
+
+ // FileSink cannot be simply cloned - it requires some special processing.
+ // Sub-queries for the union will be processed as independent map-reduce jobs
+ // possibly running in parallel. Those sub-queries cannot write to the same
+ // directory. Clone the filesink, but create a sub-directory in the final path
+ // for each sub-query. Also, these different filesinks need to be linked to each other
+ FileSinkOperator fileSinkOp = (FileSinkOperator)stack.get(pos);
+ // For file sink operator, change the directory name
+ String parentDirName = fileSinkOp.getConf().getDirName();
+
+ // Clone the fileSinkDesc of the final fileSink and create similar fileSinks at
+ // each parent
+ List<FileSinkDesc> fileDescLists = new ArrayList<FileSinkDesc>();
+
+ for (Operator<? extends OperatorDesc> parent : parents) {
+ FileSinkDesc fileSinkDesc = (FileSinkDesc) fileSinkOp.getConf().clone();
+
+ String dirName = parentDirName + Path.SEPARATOR + parent.getIdentifier() ;
+ fileSinkDesc.setDirName(dirName);
+ fileSinkDesc.setLinkedFileSink(true);
+ fileSinkDesc.setParentDir(parentDirName);
+ parent.setChildOperators(null);
+ Operator<? extends OperatorDesc> tmpFileSinkOp =
+ OperatorFactory.getAndMakeChild(fileSinkDesc, parent.getSchema(), parent);
+ tmpFileSinkOp.setChildOperators(null);
+ fileDescLists.add(fileSinkDesc);
+ }
+
+ for (FileSinkDesc fileDesc : fileDescLists) {
+ fileDesc.setLinkedFileSinkDesc(fileDescLists);
+ }
+
+ // delink union
+ union.setChildOperators(null);
+ union.setParentOperators(null);
+ } catch (Exception e) {
+ throw new SemanticException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object process(Node nd, Stack<Node> stack, NodeProcessorCtx procCtx,
+ Object... nodeOutputs) throws SemanticException {
+ FileSinkOperator fileSinkOp = (FileSinkOperator)nd;
+
+ // Has this filesink already been processed
+ if (fileSinkOp.getConf().isLinkedFileSink()) {
+ return null;
+ }
+
+ int size = stack.size();
+ int pos = size - 2;
+ UnionOperator union = null;
+
+ // Walk the tree. As long as the operators between the union and the filesink
+ // do not involve a reducer, and they can be pushed above the union, it makes
+ // sense to push them above the union, and remove the union. An interface
+ // has been added to the operator 'supportUnionRemoveOptimization' to denote whether
+ // this operator can be removed.
+ while (pos >= 0) {
+ Operator<? extends OperatorDesc> operator =
+ (Operator<? extends OperatorDesc>)stack.get(pos);
+
+ // Break if it encountered a union
+ if (operator instanceof UnionOperator) {
+ union = (UnionOperator)operator;
+ // No need for this optimization in case of multi-table inserts
+ if (union.getChildOperators().size() > 1) {
+ return null;
+ }
+ break;
+ }
+
+ if (!operator.supportUnionRemoveOptimization()) {
+ return null;
+ }
+ pos--;
+ }
+
+ UnionProcContext ctx = (UnionProcContext) procCtx;
+ UnionParseContext uCtx = ctx.getUnionParseContext(union);
+
+ // No need for this if all sub-queries are map-only queries
+ // If all the queries are map-only, anyway the query is most optimized
+ if ((uCtx != null) && (uCtx.allMapOnlySubQ())) {
+ return null;
+ }
+
+ pos++;
+ pushOperatorsAboveUnion(union, stack, pos);
+ return null;
+ }
+ }
+
+ /**
* Default processor.
*/
public static class NoUnion implements NodeProcessor {
@@ -204,4 +342,7 @@ public final class UnionProcFactory {
return new NoUnion();
}
+ public static NodeProcessor getUnionNoProcessFile() {
+ return new UnionNoProcessFile();
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcessor.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcessor.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcessor.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcessor.java Tue Oct 30 23:39:17 2012
@@ -22,11 +22,13 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Map;
-import org.apache.hadoop.hive.ql.exec.Operator;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
import org.apache.hadoop.hive.ql.exec.MapJoinOperator;
+import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.exec.TableScanOperator;
import org.apache.hadoop.hive.ql.exec.UnionOperator;
-import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator;
import org.apache.hadoop.hive.ql.lib.DefaultRuleDispatcher;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.GraphWalker;
@@ -77,13 +79,14 @@ public class UnionProcessor implements T
opRules.put(new RuleRegExp("R3",
TableScanOperator.getOperatorName() + "%.*" + UnionOperator.getOperatorName() + "%"),
UnionProcFactory.getMapUnion());
- opRules.put(new RuleRegExp("R3",
+ opRules.put(new RuleRegExp("R4",
MapJoinOperator.getOperatorName() + "%.*" + UnionOperator.getOperatorName() + "%"),
UnionProcFactory.getMapJoinUnion());
// The dispatcher fires the processor for the matching rule and passes the
// context along
UnionProcContext uCtx = new UnionProcContext();
+ uCtx.setParseContext(pCtx);
Dispatcher disp = new DefaultRuleDispatcher(UnionProcFactory.getNoUnion(),
opRules, uCtx);
GraphWalker ogw = new PreOrderWalker(disp);
@@ -94,6 +97,29 @@ public class UnionProcessor implements T
ogw.startWalking(topNodes, null);
pCtx.setUCtx(uCtx);
+ // Walk the tree again to see if the union can be removed completely
+ HiveConf conf = pCtx.getConf();
+ opRules.clear();
+ if (conf.getBoolVar(HiveConf.ConfVars.HIVE_OPTIMIZE_UNION_REMOVE)) {
+
+ if (!conf.getBoolVar(HiveConf.ConfVars.HIVE_HADOOP_SUPPORTS_SUBDIRECTORIES)) {
+ throw new
+ SemanticException(ErrorMsg.HIVE_UNION_REMOVE_OPTIMIZATION_NEEDS_SUBDIRECTORIES.getMsg());
+ }
+
+ opRules.put(new RuleRegExp("R5", UnionOperator.getOperatorName() + "%" +
+ ".*" + FileSinkOperator.getOperatorName() + "%"),
+ UnionProcFactory.getUnionNoProcessFile());
+
+ disp = new DefaultRuleDispatcher(UnionProcFactory.getNoUnion(), opRules, uCtx);
+ ogw = new PreOrderWalker(disp);
+
+ // Create a list of topop nodes
+ topNodes.clear();
+ topNodes.addAll(pCtx.getTopOps().values());
+ ogw.startWalking(topNodes, null);
+ }
+
return pCtx;
}
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/ConditionalResolverMergeFiles.java Tue Oct 30 23:39:17 2012
@@ -258,6 +258,53 @@ public class ConditionalResolverMergeFil
work.setMinSplitSizePerRack(targetSize);
}
+ private static class AverageSize {
+ private final long totalSize;
+ private final int numFiles;
+
+ public AverageSize(long totalSize, int numFiles) {
+ this.totalSize = totalSize;
+ this.numFiles = numFiles;
+ }
+
+ public long getTotalSize() {
+ return totalSize;
+ }
+
+ public int getNumFiles() {
+ return numFiles;
+ }
+ }
+
+ private AverageSize getAverageSize(FileSystem inpFs, Path dirPath) {
+ AverageSize dummy = new AverageSize(0, 0);
+ AverageSize error = new AverageSize(-1, -1);
+ try {
+ FileStatus[] fStats = inpFs.listStatus(dirPath);
+
+ long totalSz = 0;
+ int numFiles = 0;
+ for (FileStatus fStat : fStats) {
+ if (fStat.isDir()) {
+ AverageSize avgSzDir = getAverageSize(inpFs, fStat.getPath());
+ if (avgSzDir.getTotalSize() < 0) {
+ return error;
+ }
+ totalSz += avgSzDir.getTotalSize();
+ numFiles += avgSzDir.getNumFiles();
+ }
+ else {
+ totalSz += fStat.getLen();
+ numFiles++;
+ }
+ }
+
+ return new AverageSize(totalSz, numFiles);
+ } catch (IOException e) {
+ return error;
+ }
+ }
+
/**
* Whether to merge files inside directory given the threshold of the average file size.
*
@@ -270,23 +317,18 @@ public class ConditionalResolverMergeFil
* This could be true when the table is bucketized and all buckets are empty.
*/
private long getMergeSize(FileSystem inpFs, Path dirPath, long avgSize) {
- try {
- FileStatus[] fStats = inpFs.listStatus(dirPath);
- if (fStats.length <= 1) {
- return -1;
- }
- long totalSz = 0;
- for (FileStatus fStat : fStats) {
- totalSz += fStat.getLen();
- }
+ AverageSize averageSize = getAverageSize(inpFs, dirPath);
+ if (averageSize.getTotalSize() <= 0) {
+ return -1;
+ }
- if (totalSz < avgSize * fStats.length) {
- return totalSz;
- } else {
- return -1;
- }
- } catch (IOException e) {
+ if (averageSize.getNumFiles() <= 1) {
return -1;
}
+
+ if (averageSize.getTotalSize()/averageSize.getNumFiles() < avgSize) {
+ return averageSize.getTotalSize();
+ }
+ return -1;
}
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java Tue Oct 30 23:39:17 2012
@@ -19,6 +19,7 @@
package org.apache.hadoop.hive.ql.plan;
import java.util.ArrayList;
+import java.util.List;
import org.apache.hadoop.fs.Path;
@@ -45,6 +46,19 @@ public class FileSinkDesc extends Abstra
private DynamicPartitionCtx dpCtx;
private String staticSpec; // static partition spec ends with a '/'
private boolean gatherStats;
+
+ // This file descriptor is linked to other file descriptors.
+ // One use case is that, a union->select (star)->file sink, is broken down.
+ // For eg: consider a query like:
+ // select * from (subq1 union all subq2)x;
+ // where subq1 or subq2 involves a map-reduce job.
+ // It is broken into two independent queries involving subq1 and subq2 directly, and
+ // the sub-queries write to sub-directories of a common directory. So, the file sink
+ // descriptors for subq1 and subq2 are linked.
+ private boolean linkedFileSink = false;
+ private String parentDir;
+ transient private List<FileSinkDesc> linkedFileSinkDesc;
+
private boolean statsReliable;
public FileSinkDesc() {
@@ -79,6 +93,23 @@ public class FileSinkDesc extends Abstra
this.partitionCols = null;
}
+ @Override
+ public Object clone() throws CloneNotSupportedException {
+ FileSinkDesc ret = new FileSinkDesc(dirName, tableInfo, compressed,
+ destTableId, multiFileSpray, numFiles, totalFiles,
+ partitionCols, dpCtx);
+ ret.setCompressCodec(compressCodec);
+ ret.setCompressType(compressType);
+ ret.setGatherStats(gatherStats);
+ ret.setStaticSpec(staticSpec);
+ ret.setStatsAggPrefix(statsKeyPref);
+ ret.setLinkedFileSink(linkedFileSink);
+ ret.setParentDir(parentDir);
+ ret.setLinkedFileSinkDesc(linkedFileSinkDesc);
+ ret.setStatsReliable(statsReliable);
+ return (Object) ret;
+ }
+
@Explain(displayName = "directory", normalExplain = false)
public String getDirName() {
return dirName;
@@ -88,6 +119,10 @@ public class FileSinkDesc extends Abstra
this.dirName = dirName;
}
+ public String getFinalDirName() {
+ return linkedFileSink ? parentDir : dirName;
+ }
+
@Explain(displayName = "table")
public TableDesc getTableInfo() {
return tableInfo;
@@ -248,6 +283,22 @@ public class FileSinkDesc extends Abstra
}
}
+ public boolean isLinkedFileSink() {
+ return linkedFileSink;
+ }
+
+ public void setLinkedFileSink(boolean linkedFileSink) {
+ this.linkedFileSink = linkedFileSink;
+ }
+
+ public String getParentDir() {
+ return parentDir;
+ }
+
+ public void setParentDir(String parentDir) {
+ this.parentDir = parentDir;
+ }
+
public boolean isStatsReliable() {
return statsReliable;
}
@@ -255,4 +306,12 @@ public class FileSinkDesc extends Abstra
public void setStatsReliable(boolean statsReliable) {
this.statsReliable = statsReliable;
}
+
+ public List<FileSinkDesc> getLinkedFileSinkDesc() {
+ return linkedFileSinkDesc;
+ }
+
+ public void setLinkedFileSinkDesc(List<FileSinkDesc> linkedFileSinkDesc) {
+ this.linkedFileSinkDesc = linkedFileSinkDesc;
+ }
}
Modified: hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/SelectDesc.java
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/SelectDesc.java?rev=1403928&r1=1403927&r2=1403928&view=diff
==============================================================================
--- hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/SelectDesc.java (original)
+++ hive/trunk/ql/src/java/org/apache/hadoop/hive/ql/plan/SelectDesc.java Tue Oct 30 23:39:17 2012
@@ -67,8 +67,9 @@ public class SelectDesc extends Abstract
@Override
public Object clone() {
SelectDesc ret = new SelectDesc();
- ret.setColList(new ArrayList<ExprNodeDesc>(getColList()));
- ret.setOutputColumnNames(new ArrayList<String>(getOutputColumnNames()));
+ ret.setColList(getColList() == null ? null : new ArrayList<ExprNodeDesc>(getColList()));
+ ret.setOutputColumnNames(getOutputColumnNames() == null ? null :
+ new ArrayList<String>(getOutputColumnNames()));
ret.setSelectStar(selectStar);
ret.setSelStarNoCompute(selStarNoCompute);
return ret;
Added: hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_1.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_1.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_1.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_1.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,68 @@
+set hive.internal.ddl.list.bucketing.enable=true;
+set hive.optimize.skewjoin.compiletime = true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink and skewjoin optimization
+-- Union of 2 map-reduce subqueries is performed for the skew join
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output, it might be easier to run the test
+-- only on hadoop 23
+
+CREATE TABLE T1(key STRING, val STRING)
+SKEWED BY (key) ON ((2)) STORED AS TEXTFILE;
+
+LOAD DATA LOCAL INPATH '../data/files/T1.txt' INTO TABLE T1;
+
+CREATE TABLE T2(key STRING, val STRING)
+SKEWED BY (key) ON ((3)) STORED AS TEXTFILE;
+
+LOAD DATA LOCAL INPATH '../data/files/T2.txt' INTO TABLE T2;
+
+-- a simple join query with skew on both the tables on the join key
+
+EXPLAIN
+SELECT * FROM T1 a JOIN T2 b ON a.key = b.key;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+
+SELECT * FROM T1 a JOIN T2 b ON a.key = b.key
+ORDER BY a.key, b.key, a.val, b.val;
+
+-- test outer joins also
+
+EXPLAIN
+SELECT a.*, b.* FROM T1 a RIGHT OUTER JOIN T2 b ON a.key = b.key;
+
+SELECT a.*, b.* FROM T1 a RIGHT OUTER JOIN T2 b ON a.key = b.key
+ORDER BY a.key, b.key, a.val, b.val;
+
+create table DEST1(key1 STRING, val1 STRING, key2 STRING, val2 STRING);
+
+EXPLAIN
+INSERT OVERWRITE TABLE DEST1
+SELECT * FROM T1 a JOIN T2 b ON a.key = b.key;
+
+INSERT OVERWRITE TABLE DEST1
+SELECT * FROM T1 a JOIN T2 b ON a.key = b.key;
+
+SELECT * FROM DEST1
+ORDER BY key1, key2, val1, val2;
+
+EXPLAIN
+INSERT OVERWRITE TABLE DEST1
+SELECT * FROM T1 a RIGHT OUTER JOIN T2 b ON a.key = b.key;
+
+INSERT OVERWRITE TABLE DEST1
+SELECT * FROM T1 a RIGHT OUTER JOIN T2 b ON a.key = b.key;
+
+SELECT * FROM DEST1
+ORDER BY key1, key2, val1, val2;
Added: hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_2.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_2.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_2.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/skewjoin_union_remove_2.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,40 @@
+set hive.internal.ddl.list.bucketing.enable=true;
+set hive.optimize.skewjoin.compiletime = true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+set mapred.input.dir.recursive=true;
+
+CREATE TABLE T1(key STRING, val STRING)
+SKEWED BY (key) ON ((2), (8)) STORED AS TEXTFILE;
+
+LOAD DATA LOCAL INPATH '../data/files/T1.txt' INTO TABLE T1;
+
+CREATE TABLE T2(key STRING, val STRING)
+SKEWED BY (key) ON ((3), (8)) STORED AS TEXTFILE;
+
+LOAD DATA LOCAL INPATH '../data/files/T2.txt' INTO TABLE T2;
+
+CREATE TABLE T3(key STRING, val STRING) STORED AS TEXTFILE;
+
+LOAD DATA LOCAL INPATH '../data/files/T3.txt' INTO TABLE T3;
+
+-- This is to test the union->selectstar->filesink and skewjoin optimization
+-- Union of 3 map-reduce subqueries is performed for the skew join
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table, it might be easier
+-- to run the test only on hadoop 23
+
+EXPLAIN
+SELECT a.*, b.*, c.* FROM T1 a JOIN T2 b ON a.key = b.key JOIN T3 c on a.key = c.key;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+
+SELECT a.*, b.*, c.* FROM T1 a JOIN T2 b ON a.key = b.key JOIN T3 c on a.key = c.key
+ORDER BY a.key, b.key, c.key, a.val, b.val, c.val;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_1.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_1.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_1.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_1.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,44 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 map-reduce subqueries is performed followed by select star and a file sink
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- off
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as textfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT *
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a;
+
+insert overwrite table outputTbl1
+SELECT *
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a;
+
+desc formatted outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_10.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_10.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_10.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_10.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,57 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+set hive.merge.smallfiles.avgsize=1;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 subqueries is performed (one of which is a map-only query, and the
+-- other one contains a nested union where one of the sub-queries requires a map-reduce
+-- job), followed by select star and a file sink.
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The outer union can be removed completely.
+-- The final file format is different from the input and intermediate file format.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- on
+
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select * FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, 2 as values from inputTbl1
+) a
+)b;
+
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select * FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, 2 as values from inputTbl1
+) a
+)b;
+
+desc formatted outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_11.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_11.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_11.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_11.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,57 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+set hive.merge.smallfiles.avgsize=1;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 subqueries is performed (one of which is a map-only query, and the
+-- other one contains a nested union where also contains map only sub-queries),
+-- followed by select star and a file sink.
+-- There is no need for the union optimization, since the whole query can be performed
+-- in a single map-only job
+-- The final file format is different from the input and intermediate file format.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- on
+
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select * FROM (
+ SELECT key, 2 values from inputTbl1
+ UNION ALL
+ SELECT key, 3 as values from inputTbl1
+) a
+)b;
+
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select * FROM (
+ SELECT key, 2 as values from inputTbl1
+ UNION ALL
+ SELECT key, 3 as values from inputTbl1
+) a
+)b;
+
+desc formatted outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_12.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_12.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_12.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_12.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,50 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+set hive.merge.smallfiles.avgsize=1;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 subqueries is performed (one of which is a map-only query, and the
+-- other one is a map-join query), followed by select star and a file sink.
+-- The union optimization is applied, and the union is removed.
+
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+-- The final file format is different from the input and intermediate file format.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- on
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select /*+ mapjoin(a) */ a.key as key, b.val as values
+FROM inputTbl1 a join inputTbl1 b on a.key=b.key
+)c;
+
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select /*+ mapjoin(a) */ a.key as key, b.val as values
+FROM inputTbl1 a join inputTbl1 b on a.key=b.key
+)c;
+
+desc formatted outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_13.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_13.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_13.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_13.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,50 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+set hive.merge.smallfiles.avgsize=1;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 subqueries is performed (one of which is a mapred query, and the
+-- other one is a map-join query), followed by select star and a file sink.
+-- The union selectstar optimization should be performed, and the union should be removed.
+
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+-- The final file format is different from the input and intermediate file format.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- on
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, count(1) as values from inputTbl1 group by key
+union all
+select /*+ mapjoin(a) */ a.key as key, b.val as values
+FROM inputTbl1 a join inputTbl1 b on a.key=b.key
+)c;
+
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, count(1) as values from inputTbl1 group by key
+union all
+select /*+ mapjoin(a) */ a.key as key, b.val as values
+FROM inputTbl1 a join inputTbl1 b on a.key=b.key
+)c;
+
+desc formatted outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_14.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_14.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_14.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_14.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,52 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+set hive.auto.convert.join=true;
+set hive.merge.smallfiles.avgsize=1;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 subqueries is performed (one of which is a map-only query, and the
+-- other one contains a join, which should be performed as a map-join query at runtime),
+-- followed by select star and a file sink.
+-- The union selectstar optimization should be performed, and the union should be removed.
+
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+-- The final file format is different from the input and intermediate file format.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- on
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select a.key as key, b.val as values
+FROM inputTbl1 a join inputTbl1 b on a.key=b.key
+)c;
+
+insert overwrite table outputTbl1
+SELECT * FROM
+(
+select key, 1 as values from inputTbl1
+union all
+select a.key as key, b.val as values
+FROM inputTbl1 a join inputTbl1 b on a.key=b.key
+)c;
+
+desc formatted outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_15.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_15.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_15.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_15.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,53 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+
+set hive.exec.dynamic.partition.mode=nonstrict;
+set hive.exec.dynamic.partition=true;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 map-reduce subqueries is performed followed by select star and a file sink
+-- and the results are written to a table using dynamic partitions.
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- off
+-- This tests demonstrates that this optimization works in the presence of dynamic partitions.
+
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) partitioned by (ds string) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, count(1) as values, '1' as ds from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values, '2' as ds from inputTbl1 group by key
+) a;
+
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, count(1) as values, '1' as ds from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values, '2' as ds from inputTbl1 group by key
+) a;
+
+desc formatted outputTbl1;
+
+show partitions outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 where ds = '1' order by key, values;
+select * from outputTbl1 where ds = '2' order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_16.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_16.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_16.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_16.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,52 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=true;
+set hive.merge.mapredfiles=true;
+set hive.merge.smallfiles.avgsize=1;
+set mapred.input.dir.recursive=true;
+
+set hive.exec.dynamic.partition.mode=nonstrict;
+set hive.exec.dynamic.partition=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 map-reduce subqueries is performed followed by select star and a file sink
+-- and the results are written to a table using dynamic partitions.
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- on
+-- This test demonstrates that this optimization works in the presence of dynamic partitions.
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) partitioned by (ds string) stored as rcfile ;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, count(1) as values, '1' as ds from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values, '2' as ds from inputTbl1 group by key
+) a;
+
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, count(1) as values, '1' as ds from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values, '2' as ds from inputTbl1 group by key
+) a;
+
+desc formatted outputTbl1;
+show partitions outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 where ds = '1' order by key, values;
+select * from outputTbl1 where ds = '2' order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_17.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_17.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_17.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_17.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,49 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+
+set hive.exec.dynamic.partition.mode=nonstrict;
+set hive.exec.dynamic.partition=true;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 map-reduce subqueries is performed followed by select star and a file sink
+-- and the results are written to a table using dynamic partitions.
+-- There is no need for this optimization, since the query is a map-only query.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- off
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) partitioned by (ds string) stored as rcfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, 1 as values, '1' as ds from inputTbl1
+ UNION ALL
+ SELECT key, 2 as values, '2' as ds from inputTbl1
+) a;
+
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, 1 as values, '1' as ds from inputTbl1
+ UNION ALL
+ SELECT key, 2 as values, '2' as ds from inputTbl1
+) a;
+
+desc formatted outputTbl1;
+show partitions outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 where ds = '1' order by key, values;
+select * from outputTbl1 where ds = '2' order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_18.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_18.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_18.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_18.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,53 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+
+set hive.exec.dynamic.partition.mode=nonstrict;
+set hive.exec.dynamic.partition=true;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 map-reduce subqueries is performed followed by select star and a file sink
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- off
+-- This test demonstrates that the optimization works with dynamic partitions irrespective of the
+-- file format of the output file
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, ds string) stored as textfile;
+create table outputTbl1(key string, values bigint) partitioned by (ds string) stored as textfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, count(1) as values, ds from inputTbl1 group by key, ds
+ UNION ALL
+ SELECT key, count(1) as values, ds from inputTbl1 group by key, ds
+) a;
+
+insert overwrite table outputTbl1 partition (ds)
+SELECT *
+FROM (
+ SELECT key, count(1) as values, ds from inputTbl1 group by key, ds
+ UNION ALL
+ SELECT key, count(1) as values, ds from inputTbl1 group by key, ds
+) a;
+
+desc formatted outputTbl1;
+
+show partitions outputTbl1;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 where ds = '11' order by key, values;
+select * from outputTbl1 where ds = '18' order by key, values;
+select * from outputTbl1 where ds is not null order by key, values;
Added: hive/trunk/ql/src/test/queries/clientpositive/union_remove_19.q
URL: http://svn.apache.org/viewvc/hive/trunk/ql/src/test/queries/clientpositive/union_remove_19.q?rev=1403928&view=auto
==============================================================================
--- hive/trunk/ql/src/test/queries/clientpositive/union_remove_19.q (added)
+++ hive/trunk/ql/src/test/queries/clientpositive/union_remove_19.q Tue Oct 30 23:39:17 2012
@@ -0,0 +1,90 @@
+set hive.stats.autogather=false;
+set hive.optimize.union.remove=true;
+set hive.mapred.supports.subdirectories=true;
+
+set hive.merge.mapfiles=false;
+set hive.merge.mapredfiles=false;
+set mapred.input.dir.recursive=true;
+
+-- This is to test the union->selectstar->filesink optimization
+-- Union of 2 map-reduce subqueries is performed followed by select star and a file sink
+-- There is no need to write the temporary results of the sub-queries, and then read them
+-- again to process the union. The union can be removed completely.
+-- It does not matter, whether the output is merged or not. In this case, merging is turned
+-- off
+-- INCLUDE_HADOOP_MAJOR_VERSIONS(0.23)
+-- Since this test creates sub-directories for the output table outputTbl1, it might be easier
+-- to run the test only on hadoop 23
+
+create table inputTbl1(key string, val string) stored as textfile;
+create table outputTbl1(key string, values bigint) stored as textfile;
+
+load data local inpath '../data/files/T1.txt' into table inputTbl1;
+
+explain
+insert overwrite table outputTbl1
+SELECT a.key, a.values
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a;
+
+insert overwrite table outputTbl1
+SELECT a.key, a.values
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a;
+
+desc formatted outputTbl1;
+
+select * from outputTbl1;
+
+-- filter should be fine
+explain
+insert overwrite table outputTbl1
+SELECT a.key, a.values
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a where a.key = 7;
+
+insert overwrite table outputTbl1
+SELECT a.key, a.values
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a where a.key = 7;
+
+select * from outputTbl1;
+
+-- filters and sub-queries should be fine
+explain
+insert overwrite table outputTbl1
+select key, values from
+(
+SELECT a.key + a.key as key, a.values
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a
+) b where b.key >= 7;
+
+insert overwrite table outputTbl1
+select key, values from
+(
+SELECT a.key + a.key as key, a.values
+FROM (
+ SELECT key, count(1) as values from inputTbl1 group by key
+ UNION ALL
+ SELECT key, count(1) as values from inputTbl1 group by key
+) a
+) b where b.key >= 7;
+
+set hive.input.format=org.apache.hadoop.hive.ql.io.HiveInputFormat;
+select * from outputTbl1 order by key, values;