diff --git ql/src/java/org/apache/hadoop/hive/ql/Context.java ql/src/java/org/apache/hadoop/hive/ql/Context.java index 7b861ed..6caa312 100644 --- ql/src/java/org/apache/hadoop/hive/ql/Context.java +++ ql/src/java/org/apache/hadoop/hive/ql/Context.java @@ -359,9 +359,9 @@ public String getLocalTmpFileURI() { * external URI to which the tmp data has to be eventually moved * @return next available tmp path on the file system corresponding extURI */ - public String getExternalTmpFileURI(URI extURI) { - return getExternalScratchDir(extURI) + Path.SEPARATOR + EXT_PREFIX + - nextPathId(); + public Path getExternalTmpFileURI(URI extURI) { + return new Path(getExternalScratchDir(extURI), EXT_PREFIX + + nextPathId()); } /** diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java index 328c14b..c7d6ed8 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/DDLTask.java @@ -1538,9 +1538,7 @@ private int unarchive(Hive db, AlterTableSimpleDesc simpleDesc) throw new HiveException("Haven't found any archive where it should be"); } - Path tmpPath = new Path(driverContext - .getCtx() - .getExternalTmpFileURI(originalDir.toUri())); + Path tmpPath = driverContext.getCtx().getExternalTmpFileURI(originalDir.toUri()); try { fs = tmpPath.getFileSystem(conf); @@ -1548,11 +1546,6 @@ private int unarchive(Hive db, AlterTableSimpleDesc simpleDesc) throw new HiveException(e); } - // Some sanity checks - if (originalDir == null) { - throw new HiveException("Missing archive data in the partition"); - } - // Clarification of terms: // - The originalDir directory represents the original directory of the // partitions' files. They now contain an archived version of those files diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java index 8e0e9b7..a314ce7 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/FileSinkOperator.java @@ -253,13 +253,13 @@ private void initializeSpecPath() { // The movetask that follows subQ1 and subQ2 tasks still moves the directory // 'Parent' if ((!conf.isLinkedFileSink()) || (dpCtx == null)) { - specPath = new Path(conf.getDirName()); + specPath = conf.getDirName(); childSpecPathDynLinkedPartitions = null; return; } - specPath = new Path(conf.getParentDir()); - childSpecPathDynLinkedPartitions = Utilities.getFileNameFromDirName(conf.getDirName()); + specPath = conf.getParentDir(); + childSpecPathDynLinkedPartitions = conf.getDirName().getName(); } @Override @@ -816,10 +816,10 @@ public void jobCloseOp(Configuration hconf, boolean success) throws HiveException { try { if ((conf != null) && isNativeTable) { - String specPath = conf.getDirName(); + String specPath = conf.getDirName().toString(); DynamicPartitionCtx dpCtx = conf.getDynPartCtx(); if (conf.isLinkedFileSink() && (dpCtx != null)) { - specPath = conf.getParentDir(); + specPath = conf.getParentDir().toString(); } Utilities.mvFileToFinalPath(specPath, hconf, success, LOG, dpCtx, conf, reporter); diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java index 7dc3d59..133295b 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java @@ -2263,7 +2263,7 @@ private static void getMRTasks(List> tasks, List topOp, FileSinkDesc fsDesc) { ArrayList aliases = new ArrayList(); - String inputDir = fsDesc.getFinalDirName(); + String inputDir = fsDesc.getFinalDirName().toString(); TableDesc tblDesc = fsDesc.getTableInfo(); aliases.add(inputDir); // dummy alias: just use the input path @@ -1471,7 +1471,7 @@ private static MapWork createMRWorkForMergingFiles (HiveConf conf, public static MapWork createRCFileMergeTask(FileSinkDesc fsInputDesc, Path finalName, boolean hasDynamicPartitions) throws SemanticException { - String inputDir = fsInputDesc.getFinalDirName(); + Path inputDir = fsInputDesc.getFinalDirName(); TableDesc tblDesc = fsInputDesc.getTableInfo(); if (tblDesc.getInputFileFormatClass().equals(RCFileInputFormat.class)) { @@ -1479,22 +1479,22 @@ public static MapWork createRCFileMergeTask(FileSinkDesc fsInputDesc, ArrayList inputDirstr = new ArrayList(1); if (!hasDynamicPartitions && !GenMapRedUtils.isSkewedStoredAsDirs(fsInputDesc)) { - inputDirs.add(new Path(inputDir)); - inputDirstr.add(inputDir); + inputDirs.add(inputDir); + inputDirstr.add(inputDir.toString()); } MergeWork work = new MergeWork(inputDirs, finalName, hasDynamicPartitions, fsInputDesc.getDynPartCtx()); LinkedHashMap> pathToAliases = new LinkedHashMap>(); - pathToAliases.put(inputDir, (ArrayList) inputDirstr.clone()); + pathToAliases.put(inputDir.toString(), (ArrayList) inputDirstr.clone()); work.setMapperCannotSpanPartns(true); work.setPathToAliases(pathToAliases); work.setAliasToWork( new LinkedHashMap>()); if (hasDynamicPartitions || GenMapRedUtils.isSkewedStoredAsDirs(fsInputDesc)) { - work.getPathToPartitionInfo().put(inputDir, + work.getPathToPartitionInfo().put(inputDir.toString(), new PartitionDesc(tblDesc, null)); } work.setListBucketingCtx(fsInputDesc.getLbCtx()); @@ -1591,7 +1591,7 @@ public static boolean isSkewedStoredAsDirs(FileSinkDesc fsInputDesc) { } if ((srcDir != null) - && (srcDir.equals(new Path(fsOp.getConf().getFinalDirName())))) { + && (srcDir.equals(fsOp.getConf().getFinalDirName()))) { return mvTsk; } } @@ -1672,20 +1672,19 @@ public static Path createMoveTask(Task currTask, boolean Path dest = null; if (chDir) { - dest = new Path(fsOp.getConf().getFinalDirName()); + dest = fsOp.getConf().getFinalDirName(); // generate the temporary file // it must be on the same file system as the current destination Context baseCtx = parseCtx.getContext(); - String tmpDir = baseCtx.getExternalTmpFileURI(dest.toUri()); + Path tmpDir = baseCtx.getExternalTmpFileURI(dest.toUri()); 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); + fsConf.setDirName(new Path(tmpDir, fsConf.getDirName().getName())); } } else { fileSinkDesc.setDirName(tmpDir); diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchAggregation.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchAggregation.java index 476af4b..084f9f8 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchAggregation.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/SimpleFetchAggregation.java @@ -24,6 +24,7 @@ import java.util.Map; 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.GroupByOperator; import org.apache.hadoop.hive.ql.exec.Operator; @@ -94,7 +95,7 @@ public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx, } GroupByOperator pGBY = (GroupByOperator) stack.get(stack.size() - 5); - String fileName = FS.getConf().getFinalDirName(); + Path fileName = FS.getConf().getFinalDirName(); TableDesc tsDesc = createIntermediateFS(pGBY, fileName); for (AggregationDesc aggregation : cGBY.getConf().getAggregators()) { @@ -112,7 +113,7 @@ public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx, return null; } - private TableDesc createIntermediateFS(Operator parent, String fileName) { + private TableDesc createIntermediateFS(Operator parent, Path fileName) { TableDesc tsDesc = PlanUtils.getIntermediateFileTableDesc(PlanUtils .getFieldSchemasFromRowSchema(parent.getSchema(), "temporarycol")); diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java index e0f128b..eac0edd 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/BucketingSortingOpProcFactory.java @@ -462,7 +462,7 @@ public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx, // Set the inferred bucket columns for the file this FileSink produces if (bucketCols != null) { List newBucketCols = getNewBucketCols(bucketCols, colInfos); - bctx.getBucketedColsByDirectory().put(fop.getConf().getDirName(), newBucketCols); + bctx.getBucketedColsByDirectory().put(fop.getConf().getDirName().toString(), newBucketCols); bctx.setBucketedCols(fop, newBucketCols); } @@ -471,7 +471,7 @@ public Object process(Node nd, Stack stack, NodeProcessorCtx procCtx, // Set the inferred sort columns for the file this FileSink produces if (sortCols != null) { List newSortCols = getNewSortCols(sortCols, colInfos); - bctx.getSortedColsByDirectory().put(fop.getConf().getDirName(), newSortCols); + bctx.getSortedColsByDirectory().put(fop.getConf().getDirName().toString(), newSortCols); bctx.setSortedCols(fop, newSortCols); } diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java index 0820743..efa9768 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/CommonJoinTaskDispatcher.java @@ -246,7 +246,7 @@ private void mergeMapJoinTaskIntoItsChildMapRedTask(MapRedTask mapJoinTask, Conf } // The mapJoinTaskFileSinkOperator writes to a different directory - String childMRPath = mapJoinTaskFileSinkOperator.getConf().getDirName(); + String childMRPath = mapJoinTaskFileSinkOperator.getConf().getDirName().toString(); List childMRAliases = childMapWork.getPathToAliases().get(childMRPath); if (childMRAliases == null || childMRAliases.size() != 1) { return; diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java index da13abf..a985c4f 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java @@ -212,7 +212,7 @@ private void pushOperatorsAboveUnion(UnionOperator union, // 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(); + Path parentDirName = fileSinkOp.getConf().getDirName(); // Clone the fileSinkDesc of the final fileSink and create similar fileSinks at // each parent @@ -220,9 +220,7 @@ private void pushOperatorsAboveUnion(UnionOperator union, for (Operator parent : parents) { FileSinkDesc fileSinkDesc = (FileSinkDesc) fileSinkOp.getConf().clone(); - - String dirName = parentDirName + Path.SEPARATOR + parent.getIdentifier() ; - fileSinkDesc.setDirName(dirName); + fileSinkDesc.setDirName(new Path(parentDirName, parent.getIdentifier())); fileSinkDesc.setLinkedFileSink(true); fileSinkDesc.setParentDir(parentDirName); parent.setChildOperators(null); diff --git ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java index c836612..5697748 100644 --- ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java +++ ql/src/java/org/apache/hadoop/hive/ql/parse/DDLSemanticAnalyzer.java @@ -840,9 +840,9 @@ private void analyzeTruncateTable(ASTNode ast) throws SemanticException { TableDesc tblDesc = Utilities.getTableDesc(table); // Write the output to temporary directory and move it to the final location at the end // so the operation is atomic. - String queryTmpdir = ctx.getExternalTmpFileURI(newTblPartLoc.toUri()); - truncateTblDesc.setOutputDir(new Path(queryTmpdir)); - LoadTableDesc ltd = new LoadTableDesc(new Path(queryTmpdir), tblDesc, + Path queryTmpdir = ctx.getExternalTmpFileURI(newTblPartLoc.toUri()); + truncateTblDesc.setOutputDir(queryTmpdir); + LoadTableDesc ltd = new LoadTableDesc(queryTmpdir, tblDesc, partSpec == null ? new HashMap() : partSpec); ltd.setLbCtx(lbCtx); Task moveTsk = TaskFactory.get(new MoveWork(null, null, ltd, null, false), @@ -1455,9 +1455,9 @@ private void analyzeAlterTablePartMergeFiles(ASTNode tablePartAST, ASTNode ast, ddlWork.setNeedLock(true); Task mergeTask = TaskFactory.get(ddlWork, conf); TableDesc tblDesc = Utilities.getTableDesc(tblObj); - String queryTmpdir = ctx.getExternalTmpFileURI(newTblPartLoc.toUri()); - mergeDesc.setOutputDir(new Path(queryTmpdir)); - LoadTableDesc ltd = new LoadTableDesc(new Path(queryTmpdir), tblDesc, + Path queryTmpdir = ctx.getExternalTmpFileURI(newTblPartLoc.toUri()); + mergeDesc.setOutputDir(queryTmpdir); + LoadTableDesc ltd = new LoadTableDesc(queryTmpdir, tblDesc, partSpec == null ? new HashMap() : partSpec); ltd.setLbCtx(lbCtx); Task moveTsk = TaskFactory.get(new MoveWork(null, null, ltd, null, false), diff --git ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java index 9914b1d..09e90e8 100644 --- ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java +++ ql/src/java/org/apache/hadoop/hive/ql/parse/ImportSemanticAnalyzer.java @@ -281,7 +281,7 @@ public void analyzeInternal(ASTNode ast) throws SemanticException { private Task loadTable(URI fromURI, Table table) { Path dataPath = new Path(fromURI.toString(), "data"); - Path tmpPath = new Path(ctx.getExternalTmpFileURI(fromURI)); + Path tmpPath = ctx.getExternalTmpFileURI(fromURI); Task copyTask = TaskFactory.get(new CopyWork(dataPath, tmpPath, false), conf); LoadTableDesc loadTableWork = new LoadTableDesc(tmpPath, @@ -325,7 +325,7 @@ public void analyzeInternal(ASTNode ast) throws SemanticException { LOG.debug("adding dependent CopyWork/AddPart/MoveWork for partition " + partSpecToString(addPartitionDesc.getPartSpec()) + " with source location: " + srcLocation); - Path tmpPath = new Path(ctx.getExternalTmpFileURI(fromURI)); + Path tmpPath = ctx.getExternalTmpFileURI(fromURI); Task copyTask = TaskFactory.get(new CopyWork(new Path(srcLocation), tmpPath, false), conf); Task addPartTask = TaskFactory.get(new DDLWork(getInputs(), diff --git ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java index 344bb42..8df0854 100644 --- ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java +++ ql/src/java/org/apache/hadoop/hive/ql/parse/LoadSemanticAnalyzer.java @@ -226,7 +226,7 @@ public void analyzeInternal(ASTNode ast) throws SemanticException { // might seem redundant in the case // that the hive warehouse is also located in the local file system - but // that's just a test case. - String copyURIStr = ctx.getExternalTmpFileURI(toURI); + String copyURIStr = ctx.getExternalTmpFileURI(toURI).toString(); URI copyURI = URI.create(copyURIStr); rTask = TaskFactory.get(new CopyWork(new Path(fromURI), new Path(copyURI)), conf); fromURI = copyURI; @@ -234,7 +234,6 @@ public void analyzeInternal(ASTNode ast) throws SemanticException { // create final load/move work - String loadTmpPath = ctx.getExternalTmpFileURI(toURI); Map partSpec = ts.getPartSpec(); if (partSpec == null) { partSpec = new LinkedHashMap(); diff --git ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java index 848e0ab..985ae60 100644 --- ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java +++ ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java @@ -1264,7 +1264,7 @@ public void getMetaData(QB qb, ReadEntity parentInput) throws SemanticException } try { fname = ctx.getExternalTmpFileURI( - FileUtils.makeQualified(location, conf).toUri()); + FileUtils.makeQualified(location, conf).toUri()).toString(); } catch (Exception e) { throw new SemanticException(generateErrorMessage(ast, "Error creating temporary folder on: " + location.toString()), e); @@ -5264,7 +5264,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) Table dest_tab = null; // destination table if any Partition dest_part = null;// destination partition if any - String queryTmpdir = null; // the intermediate destination directory + Path queryTmpdir = null; // the intermediate destination directory Path dest_path = null; // the final destination directory TableDesc table_desc = null; int currentTableId = 0; @@ -5330,7 +5330,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) boolean isNonNativeTable = dest_tab.isNonNative(); if (isNonNativeTable) { - queryTmpdir = dest_path.toUri().getPath(); + queryTmpdir = dest_path; } else { queryTmpdir = ctx.getExternalTmpFileURI(dest_path.toUri()); } @@ -5355,7 +5355,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) // Create the work for moving the table // NOTE: specify Dynamic partitions in dest_tab for WriteEntity if (!isNonNativeTable) { - ltd = new LoadTableDesc(new Path(queryTmpdir),table_desc, dpCtx); + ltd = new LoadTableDesc(queryTmpdir,table_desc, dpCtx); ltd.setReplace(!qb.getParseInfo().isInsertIntoTable(dest_tab.getDbName(), dest_tab.getTableName())); ltd.setLbCtx(lbCtx); @@ -5438,7 +5438,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) lbCtx = constructListBucketingCtx(dest_part.getSkewedColNames(), dest_part.getSkewedColValues(), dest_part.getSkewedColValueLocationMaps(), dest_part.isStoredAsSubDirectories(), conf); - ltd = new LoadTableDesc(new Path(queryTmpdir), table_desc, dest_part.getSpec()); + ltd = new LoadTableDesc(queryTmpdir, table_desc, dest_part.getSpec()); ltd.setReplace(!qb.getParseInfo().isInsertIntoTable(dest_tab.getDbName(), dest_tab.getTableName())); ltd.setLbCtx(lbCtx); @@ -5473,7 +5473,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) if (isLocal) { // for local directory - we always write to map-red intermediate // store and then copy to local fs - queryTmpdir = ctx.getMRTmpFileURI(); + queryTmpdir = new Path(ctx.getMRTmpFileURI()); } else { // otherwise write to the file system implied by the directory // no copy is required. we may want to revisit this policy in future @@ -5558,7 +5558,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) } boolean isDfsDir = (dest_type.intValue() == QBMetaData.DEST_DFS_FILE); - loadFileWork.add(new LoadFileDesc(tblDesc, new Path(queryTmpdir), dest_path, isDfsDir, cols, + loadFileWork.add(new LoadFileDesc(tblDesc, queryTmpdir, dest_path, isDfsDir, cols, colTypes)); if (tblDesc == null) { @@ -5638,7 +5638,7 @@ private Operator genFileSinkPlan(String dest, QB qb, Operator input) // the same as directory name. The directory name // can be changed in the optimizer but the key should not be changed // it should be the same as the MoveWork's sourceDir. - fileSinkDesc.setStatsAggPrefix(fileSinkDesc.getDirName()); + fileSinkDesc.setStatsAggPrefix(fileSinkDesc.getDirName().toString()); if (dest_part != null) { try { diff --git ql/src/java/org/apache/hadoop/hive/ql/plan/DynamicPartitionCtx.java ql/src/java/org/apache/hadoop/hive/ql/plan/DynamicPartitionCtx.java index 29cfe9d..24db7d0 100644 --- ql/src/java/org/apache/hadoop/hive/ql/plan/DynamicPartitionCtx.java +++ ql/src/java/org/apache/hadoop/hive/ql/plan/DynamicPartitionCtx.java @@ -24,6 +24,7 @@ import java.util.List; import java.util.Map; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.metastore.Warehouse; import org.apache.hadoop.hive.ql.exec.ColumnInfo; import org.apache.hadoop.hive.ql.metadata.Table; @@ -39,7 +40,7 @@ private int numDPCols; // number of dynamic partition columns private int numSPCols; // number of static partition columns private String spPath; // path name corresponding to SP columns - private String rootPath; // the root path DP columns paths start from + private Path rootPath; // the root path DP columns paths start from private int numBuckets; // number of buckets in each partition private Map inputToDPCols; // mapping from input column names to DP columns @@ -128,11 +129,11 @@ public int getNumBuckets() { return this.numBuckets; } - public void setRootPath(String root) { + public void setRootPath(Path root) { this.rootPath = root; } - public String getRootPath() { + public Path getRootPath() { return this.rootPath; } diff --git ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java index 30b2411..747ac85 100644 --- ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java +++ ql/src/java/org/apache/hadoop/hive/ql/plan/FileSinkDesc.java @@ -30,7 +30,7 @@ @Explain(displayName = "File Output Operator") public class FileSinkDesc extends AbstractOperatorDesc { private static final long serialVersionUID = 1L; - private String dirName; + private Path dirName; // normally statsKeyPref will be the same as dirName, but the latter // could be changed in local execution optimization private String statsKeyPref; @@ -69,7 +69,7 @@ // 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; + private Path parentDir; transient private List linkedFileSinkDesc; private boolean statsReliable; @@ -81,7 +81,7 @@ public FileSinkDesc() { } - public FileSinkDesc(final String dirName, final TableDesc tableInfo, + public FileSinkDesc(final Path dirName, final TableDesc tableInfo, final boolean compressed, final int destTableId, final boolean multiFileSpray, final boolean canBeMerged, final int numFiles, final int totalFiles, final ArrayList partitionCols, final DynamicPartitionCtx dpCtx) { @@ -98,7 +98,7 @@ public FileSinkDesc(final String dirName, final TableDesc tableInfo, this.dpCtx = dpCtx; } - public FileSinkDesc(final String dirName, final TableDesc tableInfo, + public FileSinkDesc(final Path dirName, final TableDesc tableInfo, final boolean compressed) { this.dirName = dirName; @@ -132,15 +132,15 @@ public Object clone() throws CloneNotSupportedException { } @Explain(displayName = "directory", normalExplain = false) - public String getDirName() { + public Path getDirName() { return dirName; } - public void setDirName(final String dirName) { + public void setDirName(final Path dirName) { this.dirName = dirName; } - public String getFinalDirName() { + public Path getFinalDirName() { return linkedFileSink ? parentDir : dirName; } @@ -320,11 +320,11 @@ public void setLinkedFileSink(boolean linkedFileSink) { this.linkedFileSink = linkedFileSink; } - public String getParentDir() { + public Path getParentDir() { return parentDir; } - public void setParentDir(String parentDir) { + public void setParentDir(Path parentDir) { this.parentDir = parentDir; } diff --git ql/src/test/org/apache/hadoop/hive/ql/exec/TestExecDriver.java ql/src/test/org/apache/hadoop/hive/ql/exec/TestExecDriver.java index d4ad931..5991aae 100644 --- ql/src/test/org/apache/hadoop/hive/ql/exec/TestExecDriver.java +++ ql/src/test/org/apache/hadoop/hive/ql/exec/TestExecDriver.java @@ -196,8 +196,8 @@ private FilterDesc getTestFilterDesc(String column) { @SuppressWarnings("unchecked") private void populateMapPlan1(Table src) { - Operator op2 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapplan1.out", Utilities.defaultTd, true)); + Operator op2 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapplan1.out"), Utilities.defaultTd, true)); Operator op1 = OperatorFactory.get(getTestFilterDesc("key"), op2); @@ -207,8 +207,8 @@ private void populateMapPlan1(Table src) { @SuppressWarnings("unchecked") private void populateMapPlan2(Table src) { - Operator op3 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapplan2.out", Utilities.defaultTd, false)); + Operator op3 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapplan2.out"), Utilities.defaultTd, false)); Operator op2 = OperatorFactory.get(new ScriptDesc("cat", PlanUtils.getDefaultTableDesc("" + Utilities.tabCode, "key,value"), @@ -243,8 +243,8 @@ private void populateMapRedPlan1(Table src) throws SemanticException { mr.setReduceWork(rWork); // reduce side work - Operator op3 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapredplan1.out", Utilities.defaultTd, false)); + Operator op3 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapredplan1.out"), Utilities.defaultTd, false)); Operator op2 = OperatorFactory.get(new ExtractDesc( getStringColumn(Utilities.ReduceField.VALUE.toString())), op3); @@ -273,8 +273,8 @@ private void populateMapRedPlan2(Table src) throws SemanticException { mr.setReduceWork(rWork); // reduce side work - Operator op4 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapredplan2.out", Utilities.defaultTd, false)); + Operator op4 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapredplan2.out"), Utilities.defaultTd, false)); Operator op3 = OperatorFactory.get(getTestFilterDesc("0"), op4); @@ -317,8 +317,8 @@ private void populateMapRedPlan3(Table src, Table src2) throws SemanticException rWork.getTagToValueDesc().add(op2.getConf().getValueSerializeInfo()); // reduce side work - Operator op4 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapredplan3.out", Utilities.defaultTd, false)); + Operator op4 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapredplan3.out"), Utilities.defaultTd, false)); Operator op5 = OperatorFactory.get(new SelectDesc(Utilities .makeList(new ExprNodeFieldDesc(TypeInfoFactory.stringTypeInfo, @@ -360,8 +360,8 @@ private void populateMapRedPlan4(Table src) throws SemanticException { mr.setReduceWork(rWork); // reduce side work - Operator op3 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapredplan4.out", Utilities.defaultTd, false)); + Operator op3 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapredplan4.out"), Utilities.defaultTd, false)); Operator op2 = OperatorFactory.get(new ExtractDesc( getStringColumn(Utilities.ReduceField.VALUE.toString())), op3); @@ -399,8 +399,8 @@ private void populateMapRedPlan5(Table src) throws SemanticException { rWork.getTagToValueDesc().add(op0.getConf().getValueSerializeInfo()); // reduce side work - Operator op3 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapredplan5.out", Utilities.defaultTd, false)); + Operator op3 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapredplan5.out"), Utilities.defaultTd, false)); Operator op2 = OperatorFactory.get(new ExtractDesc( getStringColumn(Utilities.ReduceField.VALUE.toString())), op3); @@ -440,8 +440,8 @@ private void populateMapRedPlan6(Table src) throws SemanticException { rWork.getTagToValueDesc().add(op1.getConf().getValueSerializeInfo()); // reduce side work - Operator op3 = OperatorFactory.get(new FileSinkDesc(tmpdir + File.separator - + "mapredplan6.out", Utilities.defaultTd, false)); + Operator op3 = OperatorFactory.get(new FileSinkDesc(new Path(tmpdir + File.separator + + "mapredplan6.out"), Utilities.defaultTd, false)); Operator op2 = OperatorFactory.get(getTestFilterDesc("0"), op3);