[CARBONDATA-1236] Support absolute path without scheme in loading
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/47a05a8a Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/47a05a8a Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/47a05a8a Branch: refs/heads/streaming_ingest Commit: 47a05a8adb027b2ac8c8cc6d83e378564c4c90b9 Parents: 8852291 Author: dhatchayani <[email protected]> Authored: Tue Jun 27 15:26:12 2017 +0530 Committer: Venkata Ramana G <[email protected]> Committed: Tue Jun 27 18:03:14 2017 +0530 ---------------------------------------------------------------------- .../org/apache/carbondata/core/util/CarbonUtil.java | 13 +++++++++---- .../apache/carbondata/core/util/CarbonUtilTest.java | 6 +++--- .../carbondata/spark/util/GlobalDictionaryUtil.scala | 9 ++++++--- 3 files changed, 18 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/47a05a8a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java index f409551..2cd3114 100644 --- a/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java +++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java @@ -696,14 +696,20 @@ public final class CarbonUtil { * @param filePath */ public static String checkAndAppendHDFSUrl(String filePath) { + if (!filePath.startsWith("/")) { + filePath = "/" + filePath; + } String currentPath = filePath; if (null != filePath && filePath.length() != 0 && FileFactory.getFileType(filePath) != FileFactory.FileType.HDFS && FileFactory.getFileType(filePath) != FileFactory.FileType.VIEWFS) { String baseDFSUrl = CarbonProperties.getInstance() .getProperty(CarbonCommonConstants.CARBON_DDL_BASE_HDFS_URL); + String dfsUrl = conf.get(FS_DEFAULT_FS); if (null != baseDFSUrl) { - String dfsUrl = conf.get(FS_DEFAULT_FS); + if (!baseDFSUrl.startsWith("/")) { + baseDFSUrl = "/" + baseDFSUrl; + } if (null != dfsUrl && (dfsUrl.startsWith(HDFS_PREFIX) || dfsUrl .startsWith(VIEWFS_PREFIX))) { baseDFSUrl = dfsUrl + baseDFSUrl; @@ -711,10 +717,9 @@ public final class CarbonUtil { if (baseDFSUrl.endsWith("/")) { baseDFSUrl = baseDFSUrl.substring(0, baseDFSUrl.length() - 1); } - if (!filePath.startsWith("/")) { - filePath = "/" + filePath; - } currentPath = baseDFSUrl + filePath; + } else { + currentPath = dfsUrl + filePath; } } return currentPath; http://git-wip-us.apache.org/repos/asf/carbondata/blob/47a05a8a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java ---------------------------------------------------------------------- diff --git a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java b/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java index de015fa..b42c57e 100644 --- a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java +++ b/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java @@ -342,7 +342,7 @@ public class CarbonUtilTest { } }; String hdfsURL = CarbonUtil.checkAndAppendHDFSUrl("../core/src/test/resources/testDatabase"); - assertEquals(hdfsURL, "BASE_URL/../core/src/test/resources/testDatabase"); + assertEquals(hdfsURL, "/BASE_URL/../core/src/test/resources/testDatabase"); } @Test public void testToCheckAndAppendHDFSUrlWithBlackSlash() { @@ -357,7 +357,7 @@ public class CarbonUtilTest { } }; String hdfsURL = CarbonUtil.checkAndAppendHDFSUrl("../core/src/test/resources/testDatabase"); - assertEquals(hdfsURL, "BASE_URL/../core/src/test/resources/testDatabase"); + assertEquals(hdfsURL, "/BASE_URL/../core/src/test/resources/testDatabase"); } @Test public void testToCheckAndAppendHDFSUrlWithNull() { @@ -372,7 +372,7 @@ public class CarbonUtilTest { } }; String hdfsURL = CarbonUtil.checkAndAppendHDFSUrl("../core/src/test/resources/testDatabase"); - assertEquals(hdfsURL, "../core/src/test/resources/testDatabase"); + assertEquals(hdfsURL, "file:////../core/src/test/resources/testDatabase"); } @Test public void testForisFileExists() { http://git-wip-us.apache.org/repos/asf/carbondata/blob/47a05a8a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala index 364476d..bf41dca 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala @@ -441,7 +441,9 @@ object GlobalDictionaryUtil { " should be columnName:columnPath, please check") } setPredefineDict(carbonLoadModel, dimensions, table, colNameWithPath(0), - FileUtils.getPaths(colPathMapTrim.substring(colNameWithPath(0).length + 1))) + FileUtils + .getPaths(CarbonUtil + .checkAndAppendHDFSUrl(colPathMapTrim.substring(colNameWithPath(0).length + 1)))) } } @@ -773,7 +775,8 @@ object GlobalDictionaryUtil { dimensions: Array[CarbonDimension], allDictionaryPath: String): Unit = { LOGGER.info("Generate global dictionary from dictionary files!") - val isNonempty = validateAllDictionaryPath(allDictionaryPath) + val allDictionaryPathAppended = CarbonUtil.checkAndAppendHDFSUrl(allDictionaryPath) + val isNonempty = validateAllDictionaryPath(allDictionaryPathAppended) if (isNonempty) { var headers = carbonLoadModel.getCsvHeaderColumns headers = headers.map(headerName => headerName.trim) @@ -786,7 +789,7 @@ object GlobalDictionaryUtil { val accumulator = sqlContext.sparkContext.accumulator(0) // read local dictionary file, and group by key val allDictionaryRdd = readAllDictionaryFiles(sqlContext, headers, - requireColumnNames, allDictionaryPath, accumulator) + requireColumnNames, allDictionaryPathAppended, accumulator) // read exist dictionary and combine val inputRDD = new CarbonAllDictionaryCombineRDD(allDictionaryRdd, model) .partitionBy(new ColumnPartitioner(model.primDimensions.length))
