diff --git ql/src/java/org/apache/hadoop/hive/ql/Context.java ql/src/java/org/apache/hadoop/hive/ql/Context.java index 0b31c42..ea59524 100644 --- ql/src/java/org/apache/hadoop/hive/ql/Context.java +++ ql/src/java/org/apache/hadoop/hive/ql/Context.java @@ -311,9 +311,9 @@ public boolean isMRTmpFileURI(String uriStr) { * * @return next available path for map-red intermediate data */ - public String getMRTmpFileURI() { - return getMRScratchDir() + Path.SEPARATOR + MR_PREFIX + - nextPathId(); + public Path getMRTmpFileURI() { + return new Path(getMRScratchDir(), MR_PREFIX + + nextPathId()); } diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java index aa8f19c..c54717b 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/HashTableSinkOperator.java @@ -266,7 +266,7 @@ public void closeOp(boolean abort) throws HiveException { try { if (mapJoinTables != null) { // get tmp file URI - String tmpURI = this.getExecContext().getLocalWork().getTmpFileURI(); + Path tmpURI = this.getExecContext().getLocalWork().getTmpFileURI(); LOG.info("Temp URI for side table: " + tmpURI); for (byte tag = 0; tag < mapJoinTables.length; tag++) { // get the key and value @@ -279,10 +279,9 @@ public void closeOp(boolean abort) throws HiveException { String fileName = getExecContext().getLocalWork().getBucketFileName(bigBucketFileName); // get the tmp URI path; it will be a hdfs path if not local mode String dumpFilePrefix = conf.getDumpFilePrefix(); - String tmpURIPath = Utilities.generatePath(tmpURI, dumpFilePrefix, tag, fileName); - console.printInfo(Utilities.now() + "\tDump the side-table into file: " + tmpURIPath); + Path path = Utilities.generatePath(tmpURI, dumpFilePrefix, tag, fileName); + console.printInfo(Utilities.now() + "\tDump the side-table into file: " + path); // get the hashtable file and path - Path path = new Path(tmpURIPath); FileSystem fs = path.getFileSystem(hconf); ObjectOutputStream out = new ObjectOutputStream(new BufferedOutputStream(fs.create(path), 4096)); try { @@ -291,7 +290,7 @@ public void closeOp(boolean abort) throws HiveException { out.close(); } tableContainer.clear(); - console.printInfo(Utilities.now() + "\tUpload 1 File to: " + tmpURIPath); + console.printInfo(Utilities.now() + "\tUpload 1 File to: " + path); } } super.closeOp(abort); 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 fccea89..a7d6c91 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/Utilities.java @@ -563,22 +563,22 @@ protected Expression instantiate(Object oldInstance, Encoder out) { } } - public static void setMapRedWork(Configuration conf, MapredWork w, String hiveScratchDir) { + public static void setMapRedWork(Configuration conf, MapredWork w, Path hiveScratchDir) { setMapWork(conf, w.getMapWork(), hiveScratchDir, true); if (w.getReduceWork() != null) { setReduceWork(conf, w.getReduceWork(), hiveScratchDir, true); } } - public static Path setMapWork(Configuration conf, MapWork w, String hiveScratchDir, boolean useCache) { + public static Path setMapWork(Configuration conf, MapWork w, Path hiveScratchDir, boolean useCache) { return setBaseWork(conf, w, hiveScratchDir, MAP_PLAN_NAME, useCache); } - public static Path setReduceWork(Configuration conf, ReduceWork w, String hiveScratchDir, boolean useCache) { + public static Path setReduceWork(Configuration conf, ReduceWork w, Path hiveScratchDir, boolean useCache) { return setBaseWork(conf, w, hiveScratchDir, REDUCE_PLAN_NAME, useCache); } - private static Path setBaseWork(Configuration conf, BaseWork w, String hiveScratchDir, String name, boolean useCache) { + private static Path setBaseWork(Configuration conf, BaseWork w, Path hiveScratchDir, String name, boolean useCache) { try { setPlanPath(conf, hiveScratchDir); @@ -635,7 +635,7 @@ private static Path getPlanPath(Configuration conf, String name) { return new Path(planPath, name); } - private static void setPlanPath(Configuration conf, String hiveScratchDir) throws IOException { + private static void setPlanPath(Configuration conf, Path hiveScratchDir) throws IOException { if (getPlanPath(conf) == null) { // this is the unique conf ID, which is kept in JobConf as part of the plan file name String jobID = UUID.randomUUID().toString(); @@ -2468,11 +2468,10 @@ public static void validatePartSpecColumnNames(Table tbl, Map pa public static String suffix = ".hashtable"; - public static String generatePath(String baseURI, String dumpFilePrefix, + public static Path generatePath(Path baseURI, String dumpFilePrefix, Byte tag, String bigBucketFileName) { - String path = new String(baseURI + Path.SEPARATOR + "MapJoin-" + dumpFilePrefix + tag + + return new Path(baseURI, "MapJoin-" + dumpFilePrefix + tag + "-" + bigBucketFileName + suffix); - return path; } public static String generateFileName(Byte tag, String bigBucketFileName) { @@ -2480,9 +2479,8 @@ public static String generateFileName(Byte tag, String bigBucketFileName) { return fileName; } - public static String generateTmpURI(String baseURI, String id) { - String tmpFileURI = new String(baseURI + Path.SEPARATOR + "HashTable-" + id); - return tmpFileURI; + public static Path generateTmpURI(Path baseURI, String id) { + return new Path(baseURI, "HashTable-" + id); } public static String generateTarURI(String baseURI, String filename) { @@ -2921,7 +2919,7 @@ public static double getHighestSamplePercentage (MapWork work) { * @return List of paths to process for the given MapWork * @throws Exception */ - public static List getInputPaths(JobConf job, MapWork work, String hiveScratchDir, Context ctx) + public static List getInputPaths(JobConf job, MapWork work, Path hiveScratchDir, Context ctx) throws Exception { int sequenceNumber = 0; @@ -2976,7 +2974,7 @@ public static double getHighestSamplePercentage (MapWork work) { } @SuppressWarnings({"rawtypes", "unchecked"}) - private static Path createEmptyFile(String hiveScratchDir, + private static Path createEmptyFile(Path hiveScratchDir, Class outFileFormat, JobConf job, int sequenceNumber, Properties props, boolean dummyRow) throws IOException, InstantiationException, IllegalAccessException { @@ -2993,7 +2991,6 @@ private static Path createEmptyFile(String hiveScratchDir, String newFile = newDir + File.separator + "emptyFile"; Path newFilePath = new Path(newFile); - String onefile = newPath.toString(); FSRecordWriter recWriter = outFileFormat.newInstance().getHiveRecordWriter(job, newFilePath, Text.class, false, props, null); if (dummyRow) { @@ -3009,7 +3006,7 @@ private static Path createEmptyFile(String hiveScratchDir, @SuppressWarnings("rawtypes") private static Path createDummyFileForEmptyPartition(Path path, JobConf job, MapWork work, - String hiveScratchDir, String alias, int sequenceNumber) + Path hiveScratchDir, String alias, int sequenceNumber) throws IOException, InstantiationException, IllegalAccessException { String strPath = path.toString(); @@ -3051,7 +3048,7 @@ private static Path createDummyFileForEmptyPartition(Path path, JobConf job, Map @SuppressWarnings("rawtypes") private static Path createDummyFileForEmptyTable(JobConf job, MapWork work, - String hiveScratchDir, String alias, int sequenceNumber) + Path hiveScratchDir, String alias, int sequenceNumber) throws IOException, InstantiationException, IllegalAccessException { TableDesc tableDesc = work.getAliasToPartnInfo().get(alias).getTableDesc(); diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java index 5511bca..d691701 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/mr/ExecDriver.java @@ -203,7 +203,6 @@ public int execute(DriverContext driverContext) { Context ctx = driverContext.getCtx(); boolean ctxCreated = false; - String emptyScratchDirStr; Path emptyScratchDir; MapWork mWork = work.getMapWork(); @@ -215,8 +214,7 @@ public int execute(DriverContext driverContext) { ctxCreated = true; } - emptyScratchDirStr = ctx.getMRTmpFileURI(); - emptyScratchDir = new Path(emptyScratchDirStr); + emptyScratchDir = ctx.getMRTmpFileURI(); FileSystem fs = emptyScratchDir.getFileSystem(job); fs.mkdirs(emptyScratchDir); } catch (IOException e) { @@ -331,8 +329,8 @@ public int execute(DriverContext driverContext) { MapredLocalWork localwork = mWork.getMapLocalWork(); if (localwork != null) { if (!ShimLoader.getHadoopShims().isLocalMode(job)) { - Path localPath = new Path(localwork.getTmpFileURI()); - Path hdfsPath = new Path(mWork.getTmpHDFSFileURI()); + Path localPath = localwork.getTmpFileURI(); + Path hdfsPath = mWork.getTmpHDFSFileURI(); FileSystem hdfs = hdfsPath.getFileSystem(job); FileSystem localFS = localPath.getFileSystem(job); @@ -370,7 +368,7 @@ public int execute(DriverContext driverContext) { } } work.configureJobConf(job); - List inputPaths = Utilities.getInputPaths(job, mWork, emptyScratchDirStr, ctx); + List inputPaths = Utilities.getInputPaths(job, mWork, emptyScratchDir, ctx); Utilities.setInputPaths(job, inputPaths); Utilities.setMapRedWork(job, work, ctx.getMRTmpFileURI()); diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HashTableLoader.java ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HashTableLoader.java index efe5710..86642c7 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HashTableLoader.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/mr/HashTableLoader.java @@ -55,7 +55,7 @@ public void load(ExecMapperContext context, MapJoinTableContainer[] mapJoinTables, MapJoinTableContainerSerDe[] mapJoinTableSerdes) throws HiveException { - String baseDir = null; + Path baseDir = null; Path currentInputPath = context.getCurrentInputPath(); LOG.info("******* Load from HashTable File: input : " + currentInputPath); String fileName = context.getLocalWork().getBucketFileName(currentInputPath.toString()); @@ -74,8 +74,7 @@ public void load(ExecMapperContext context, if (!archive.getName().endsWith(suffix)) { continue; } - Path archiveLocalLink = archive.makeQualified(localFs); - baseDir = archiveLocalLink.toUri().getPath(); + baseDir = archive.makeQualified(localFs); } } for (int pos = 0; pos < mapJoinTables.length; pos++) { @@ -85,8 +84,7 @@ public void load(ExecMapperContext context, if(baseDir == null) { throw new IllegalStateException("baseDir cannot be null"); } - String filePath = Utilities.generatePath(baseDir, desc.getDumpFilePrefix(), (byte)pos, fileName); - Path path = new Path(filePath); + Path path = Utilities.generatePath(baseDir, desc.getDumpFilePrefix(), (byte)pos, fileName); LOG.info("\tLoad back 1 hashtable file from tmp file uri:" + path); ObjectInputStream in = new ObjectInputStream(new BufferedInputStream( new FileInputStream(path.toUri().getPath()), 4096)); diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java index 0cc90d0..b3600b4 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/mr/MapredLocalTask.java @@ -427,15 +427,14 @@ private void generateDummyHashTable(String alias, String bigBucketFileName) byte tag = (byte) childOp.getParentOperators().indexOf(parentOp); // generate empty hashtable for this (byte)tag - String tmpURI = this.getWork().getTmpFileURI(); + Path tmpURI = this.getWork().getTmpFileURI(); String fileName = work.getBucketFileName(bigBucketFileName); HashTableSinkOperator htso = (HashTableSinkOperator)childOp; - String tmpURIPath = Utilities.generatePath(tmpURI, htso.getConf().getDumpFilePrefix(), + Path path = Utilities.generatePath(tmpURI, htso.getConf().getDumpFilePrefix(), tag, fileName); - console.printInfo(Utilities.now() + "\tDump the hashtable into file: " + tmpURIPath); - Path path = new Path(tmpURIPath); + console.printInfo(Utilities.now() + "\tDump the hashtable into file: " + path); FileSystem fs = path.getFileSystem(job); ObjectOutputStream out = new ObjectOutputStream(fs.create(path)); try { @@ -443,7 +442,7 @@ private void generateDummyHashTable(String alias, String bigBucketFileName) } finally { out.close(); } - console.printInfo(Utilities.now() + "\tUpload 1 File to: " + tmpURIPath + " File size: " + console.printInfo(Utilities.now() + "\tUpload 1 File to: " + path + " File size: " + fs.getFileStatus(path).getLen()); } diff --git ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java index 69563e6..883e716 100644 --- ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java +++ ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java @@ -207,12 +207,11 @@ private Vertex createVertex(JobConf conf, MapWork mapWork, Path tezDir = getTezDir(mrScratchDir); // set up the operator plan - Path planPath = Utilities.setMapWork(conf, mapWork, - mrScratchDir.toUri().toString(), false); + Utilities.setMapWork(conf, mapWork, mrScratchDir, false); // setup input paths and split info List inputPaths = Utilities.getInputPaths(conf, mapWork, - mrScratchDir.toUri().toString(), ctx); + mrScratchDir, ctx); Utilities.setInputPaths(conf, inputPaths); // create the directories FileSinkOperators need @@ -317,8 +316,7 @@ private Vertex createVertex(JobConf conf, ReduceWork reduceWork, Path mrScratchDir, Context ctx) throws Exception { // set up operator plan - Path planPath = Utilities.setReduceWork(conf, reduceWork, - mrScratchDir.toUri().toString(), false); + Utilities.setReduceWork(conf, reduceWork, mrScratchDir, false); // create the directories FileSinkOperators need Utilities.createTmpDirs(conf, reduceWork); diff --git ql/src/java/org/apache/hadoop/hive/ql/index/bitmap/BitmapIndexHandler.java ql/src/java/org/apache/hadoop/hive/ql/index/bitmap/BitmapIndexHandler.java index 34ced8b..2bbe1e6 100644 --- ql/src/java/org/apache/hadoop/hive/ql/index/bitmap/BitmapIndexHandler.java +++ ql/src/java/org/apache/hadoop/hive/ql/index/bitmap/BitmapIndexHandler.java @@ -97,7 +97,7 @@ public void generateIndexQuery(List indexes, ExprNodeDesc predicate, // Build reentrant QL for index query StringBuilder qlCommand = new StringBuilder("INSERT OVERWRITE DIRECTORY "); - String tmpFile = pctx.getContext().getMRTmpFileURI(); + String tmpFile = pctx.getContext().getMRTmpFileURI().toUri().toString(); qlCommand.append( "\"" + tmpFile + "\" "); // QL includes " around file name qlCommand.append("SELECT bucketname AS `_bucketname` , COLLECT_SET(offset) AS `_offsets` FROM "); qlCommand.append("(SELECT `_bucketname` AS bucketname , `_offset` AS offset FROM "); diff --git ql/src/java/org/apache/hadoop/hive/ql/index/compact/CompactIndexHandler.java ql/src/java/org/apache/hadoop/hive/ql/index/compact/CompactIndexHandler.java index 57ebd1e..48e7047 100644 --- ql/src/java/org/apache/hadoop/hive/ql/index/compact/CompactIndexHandler.java +++ ql/src/java/org/apache/hadoop/hive/ql/index/compact/CompactIndexHandler.java @@ -170,7 +170,7 @@ public void generateIndexQuery(List indexes, ExprNodeDesc predicate, // Build reentrant QL for index query StringBuilder qlCommand = new StringBuilder("INSERT OVERWRITE DIRECTORY "); - String tmpFile = pctx.getContext().getMRTmpFileURI(); + String tmpFile = pctx.getContext().getMRTmpFileURI().toUri().toString(); queryContext.setIndexIntermediateFile(tmpFile); qlCommand.append( "\"" + tmpFile + "\" "); // QL includes " around file name qlCommand.append("SELECT `_bucketname` , `_offsets` FROM "); diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java index c580818..bb2209e 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMRUnion1.java @@ -22,6 +22,7 @@ import java.util.Map; import java.util.Stack; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.ql.Context; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.TableScanOperator; @@ -124,7 +125,7 @@ private void processSubQueryUnionCreateIntermediate( // generate the temporary file Context baseCtx = parseCtx.getContext(); - String taskTmpDir = baseCtx.getMRTmpFileURI(); + Path taskTmpDir = baseCtx.getMRTmpFileURI(); // Create the temporary file, its corresponding FileSinkOperaotr, and // its corresponding TableScanOperator. @@ -132,7 +133,7 @@ private void processSubQueryUnionCreateIntermediate( GenMapRedUtils.createTemporaryFile(parent, child, taskTmpDir, tt_desc, parseCtx); // Add the path to alias mapping - uCtxTask.addTaskTmpDir(taskTmpDir); + uCtxTask.addTaskTmpDir(taskTmpDir.toUri().toString()); uCtxTask.addTTDesc(tt_desc); uCtxTask.addListTopOperators(tableScanOp); diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java index ac9df5e..ae95b32 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/GenMapRedUtils.java @@ -946,12 +946,12 @@ public static TableScanOperator createTemporaryTableScanOperator(RowSchema rowSc */ protected static TableScanOperator createTemporaryFile( Operator parent, Operator child, - String taskTmpDir, TableDesc tt_desc, ParseContext parseCtx) { + Path taskTmpDir, TableDesc tt_desc, ParseContext parseCtx) { // Create a FileSinkOperator for the file name of taskTmpDir boolean compressIntermediate = parseCtx.getConf().getBoolVar(HiveConf.ConfVars.COMPRESSINTERMEDIATE); - FileSinkDesc desc = new FileSinkDesc(new Path(taskTmpDir), tt_desc, compressIntermediate); + FileSinkDesc desc = new FileSinkDesc(taskTmpDir, tt_desc, compressIntermediate); if (compressIntermediate) { desc.setCompressCodec(parseCtx.getConf().getVar( HiveConf.ConfVars.COMPRESSINTERMEDIATECODEC)); @@ -1008,7 +1008,7 @@ private static void splitTasks(ReduceSinkOperator op, // Generate the temporary file name Context baseCtx = parseCtx.getContext(); - String taskTmpDir = baseCtx.getMRTmpFileURI(); + Path taskTmpDir = baseCtx.getMRTmpFileURI(); Operator parent = op.getParentOperators().get(0); TableDesc tt_desc = PlanUtils.getIntermediateFileTableDesc(PlanUtils @@ -1023,7 +1023,7 @@ private static void splitTasks(ReduceSinkOperator op, opProcCtx.getMapCurrCtx(); mapCurrCtx.put(tableScanOp, new GenMapRedCtx(childTask, null)); - String streamDesc = taskTmpDir; + String streamDesc = taskTmpDir.toUri().toString(); MapredWork cplan = (MapredWork) childTask.getWork(); if (needsTagging(cplan.getReduceWork())) { @@ -1055,7 +1055,7 @@ private static void splitTasks(ReduceSinkOperator op, } // Add the path to alias mapping - setTaskPlan(taskTmpDir, streamDesc, tableScanOp, cplan.getMapWork(), false, tt_desc); + setTaskPlan(taskTmpDir.toUri().toString(), streamDesc, tableScanOp, cplan.getMapWork(), false, tt_desc); opProcCtx.setCurrTopOp(null); opProcCtx.setCurrAliasId(null); opProcCtx.setCurrTask(childTask); diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/GenMRSkewJoinProcessor.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/GenMRSkewJoinProcessor.java index 4a775d9..5dec160 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/GenMRSkewJoinProcessor.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/GenMRSkewJoinProcessor.java @@ -123,7 +123,7 @@ public static void processSkewJoin(JoinOperator joinOp, Task child = children != null && children.size() == 1 ? children.get(0) : null; - String baseTmpDir = parseCtx.getContext().getMRTmpFileURI(); + String baseTmpDir = parseCtx.getContext().getMRTmpFileURI().toUri().toString(); JoinDesc joinDescriptor = joinOp.getConf(); Map> joinValues = joinDescriptor.getExprs(); diff --git ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java index 010ac54..e03303d 100644 --- ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java +++ ql/src/java/org/apache/hadoop/hive/ql/optimizer/physical/MapJoinResolver.java @@ -26,6 +26,7 @@ import java.util.Map; import java.util.Stack; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.ql.Context; import org.apache.hadoop.hive.ql.exec.ConditionalTask; import org.apache.hadoop.hive.ql.exec.MapJoinOperator; @@ -101,9 +102,9 @@ private void processCurrentTask(Task currTask, if (localwork != null) { // get the context info and set up the shared tmp URI Context ctx = physicalContext.getContext(); - String tmpFileURI = Utilities.generateTmpURI(ctx.getLocalTmpFileURI(), currTask.getId()); + Path tmpFileURI = Utilities.generateTmpURI(new Path(ctx.getLocalTmpFileURI()), currTask.getId()); localwork.setTmpFileURI(tmpFileURI); - String hdfsTmpURI = Utilities.generateTmpURI(ctx.getMRTmpFileURI(), currTask.getId()); + Path hdfsTmpURI = Utilities.generateTmpURI(ctx.getMRTmpFileURI(), currTask.getId()); mapredWork.getMapWork().setTmpHDFSFileURI(hdfsTmpURI); // create a task for this local work; right now, this local work is shared // by the original MapredTask and this new generated MapredLocalTask. 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 52d7c75..58cf46b 100644 --- ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java +++ ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java @@ -1278,7 +1278,7 @@ public void getMetaData(QB qb, ReadEntity parentInput) throws SemanticException } } else { qb.setIsQuery(true); - fname = ctx.getMRTmpFileURI(); + fname = ctx.getMRTmpFileURI().toString(); ctx.setResDir(new Path(fname)); } } @@ -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 = new Path(ctx.getMRTmpFileURI()); + queryTmpdir = 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 diff --git ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java index 9929275..fa31091 100644 --- ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java +++ ql/src/java/org/apache/hadoop/hive/ql/plan/MapWork.java @@ -83,7 +83,7 @@ new HashMap>(); private MapredLocalWork mapLocalWork; - private String tmpHDFSFileURI; + private Path tmpHDFSFileURI; private String inputformat; @@ -432,11 +432,11 @@ public void setOpParseCtxMap( this.opParseCtxMap = opParseCtxMap; } - public String getTmpHDFSFileURI() { + public Path getTmpHDFSFileURI() { return tmpHDFSFileURI; } - public void setTmpHDFSFileURI(String tmpHDFSFileURI) { + public void setTmpHDFSFileURI(Path tmpHDFSFileURI) { this.tmpHDFSFileURI = tmpHDFSFileURI; } diff --git ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java index 83a778d..8f7e18e 100644 --- ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java +++ ql/src/java/org/apache/hadoop/hive/ql/plan/MapredLocalWork.java @@ -37,7 +37,7 @@ private LinkedHashMap aliasToFetchWork; private boolean inputFileChangeSensitive; private BucketMapJoinContext bucketMapjoinContext; - private String tmpFileURI; + private Path tmpFileURI; private String stageID; private List> dummyParentOp ; @@ -151,11 +151,11 @@ public BucketMapJoinContext copyPartSpecMappingOnly() { return null; } - public void setTmpFileURI(String tmpFileURI) { + public void setTmpFileURI(Path tmpFileURI) { this.tmpFileURI = tmpFileURI; } - public String getTmpFileURI() { + public Path getTmpFileURI() { return tmpFileURI; } diff --git ql/src/test/org/apache/hadoop/hive/ql/exec/TestPlan.java ql/src/test/org/apache/hadoop/hive/ql/exec/TestPlan.java index f1e8555..1364888 100644 --- ql/src/test/org/apache/hadoop/hive/ql/exec/TestPlan.java +++ ql/src/test/org/apache/hadoop/hive/ql/exec/TestPlan.java @@ -25,6 +25,7 @@ import junit.framework.TestCase; +import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.ql.parse.TypeCheckProcFactory; import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc; import org.apache.hadoop.hive.ql.plan.ExprNodeDesc; @@ -89,8 +90,8 @@ public void testPlan() throws Exception { // store into configuration job.set("fs.default.name", "file:///"); - Utilities.setMapRedWork(job, mrwork, System.getProperty("java.io.tmpdir") + File.separator + - System.getProperty("user.name") + File.separator + "hive"); + Utilities.setMapRedWork(job, mrwork, new Path(System.getProperty("java.io.tmpdir") + File.separator + + System.getProperty("user.name") + File.separator + "hive")); MapredWork mrwork2 = Utilities.getMapRedWork(job); Utilities.clearWork(job); diff --git ql/src/test/org/apache/hadoop/hive/ql/io/TestHiveBinarySearchRecordReader.java ql/src/test/org/apache/hadoop/hive/ql/io/TestHiveBinarySearchRecordReader.java index 1047a8d..292a835 100644 --- ql/src/test/org/apache/hadoop/hive/ql/io/TestHiveBinarySearchRecordReader.java +++ ql/src/test/org/apache/hadoop/hive/ql/io/TestHiveBinarySearchRecordReader.java @@ -140,7 +140,7 @@ private void init() throws IOException { pt.put("/tmp/testfolder", partDesc); MapredWork mrwork = new MapredWork(); mrwork.getMapWork().setPathToPartitionInfo(pt); - Utilities.setMapRedWork(conf, mrwork,"/tmp/" + System.getProperty("user.name") + "/hive"); + Utilities.setMapRedWork(conf, mrwork,new Path("/tmp/" + System.getProperty("user.name"), "hive")); hiveSplit = new TestHiveInputSplit(); hbsReader = new TestHiveRecordReader(rcfReader, conf); diff --git ql/src/test/org/apache/hadoop/hive/ql/io/TestSymlinkTextInputFormat.java ql/src/test/org/apache/hadoop/hive/ql/io/TestSymlinkTextInputFormat.java index 94061af..b0273a6 100644 --- ql/src/test/org/apache/hadoop/hive/ql/io/TestSymlinkTextInputFormat.java +++ ql/src/test/org/apache/hadoop/hive/ql/io/TestSymlinkTextInputFormat.java @@ -81,7 +81,7 @@ protected void setUp() throws IOException { pt.put("/tmp/testfolder", partDesc); MapredWork mrwork = new MapredWork(); mrwork.getMapWork().setPathToPartitionInfo(pt); - Utilities.setMapRedWork(job, mrwork,"/tmp/" + System.getProperty("user.name") + "/hive"); + Utilities.setMapRedWork(job, mrwork,new Path("/tmp/" + System.getProperty("user.name"), "hive")); fileSystem = FileSystem.getLocal(conf); testDir = new Path(System.getProperty("test.tmp.dir", System.getProperty( @@ -169,18 +169,15 @@ public void testCombine() throws Exception { drv.compile(cmd); //create scratch dir - String emptyScratchDirStr; - Path emptyScratchDir; Context ctx = new Context(newJob); - emptyScratchDirStr = ctx.getMRTmpFileURI(); - emptyScratchDir = new Path(emptyScratchDirStr); + Path emptyScratchDir = ctx.getMRTmpFileURI(); FileSystem fileSys = emptyScratchDir.getFileSystem(newJob); fileSys.mkdirs(emptyScratchDir); QueryPlan plan = drv.getPlan(); MapRedTask selectTask = (MapRedTask)plan.getRootTasks().get(0); - List inputPaths = Utilities.getInputPaths(newJob, selectTask.getWork().getMapWork(), emptyScratchDir.toString(), ctx); + List inputPaths = Utilities.getInputPaths(newJob, selectTask.getWork().getMapWork(), emptyScratchDir, ctx); Utilities.setInputPaths(newJob, inputPaths); Utilities.setMapRedWork(newJob, selectTask.getWork(), ctx.getMRTmpFileURI());