fix testcase clean
clean clean fix comment Project: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/commit/7b8b1959 Tree: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/tree/7b8b1959 Diff: http://git-wip-us.apache.org/repos/asf/incubator-carbondata/diff/7b8b1959 Branch: refs/heads/master Commit: 7b8b19598bf61001b2c4e07debbe26868f64d95c Parents: b966043 Author: jackylk <jacky.li...@huawei.com> Authored: Fri Dec 30 14:44:14 2016 +0800 Committer: chenliang613 <chenliang...@apache.org> Committed: Wed Jan 4 17:11:54 2017 +0800 ---------------------------------------------------------------------- .../scan/executor/impl/DetailQueryExecutor.java | 13 ++- .../impl/VectorDetailQueryExecutor.java | 11 +- .../scan/executor/util/QueryUtil.java | 4 +- .../carbondata/scan/filter/FilterUtil.java | 6 +- .../AbstractDetailQueryResultIterator.java | 2 +- .../iterator/DetailQueryResultIterator.java | 2 +- .../VectorDetailQueryResultIterator.java | 2 +- .../MalformedCarbonCommandException.java | 14 --- .../carbondata/spark/load/CarbonLoaderUtil.java | 18 ++- .../spark/load/DeleteLoadFolders.java | 54 +-------- .../spark/load/DeletedLoadMetadata.java | 53 --------- .../spark/merger/CarbonCompactionExecutor.java | 35 +----- .../spark/merger/CarbonDataMergerUtil.java | 48 +------- .../spark/merger/RowResultMerger.java | 8 +- .../spark/merger/TupleConversionAdapter.java | 6 +- .../spark/partition/api/DataPartitioner.java | 19 +-- .../spark/partition/api/Partition.java | 5 - .../api/impl/DataPartitionerProperties.java | 87 -------------- .../partition/api/impl/DefaultLoadBalancer.java | 4 - .../spark/partition/api/impl/PartitionImpl.java | 54 --------- .../api/impl/PartitionMultiFileImpl.java | 5 - .../api/impl/QueryPartitionHelper.java | 3 +- .../api/impl/SampleDataPartitionerImpl.java | 117 +------------------ .../carbondata/spark/util/CarbonQueryUtil.java | 14 +-- .../apache/carbondata/spark/CarbonFilters.scala | 104 +++++++++++------ .../spark/rdd/CarbonDataLoadRDD.scala | 10 +- .../spark/rdd/CarbonDeleteLoadByDateRDD.scala | 4 +- .../spark/rdd/CarbonGlobalDictionaryRDD.scala | 36 +++--- .../carbondata/spark/rdd/CarbonMergerRDD.scala | 6 +- .../spark/rdd/DataLoadPartitionCoalescer.scala | 8 +- .../spark/rdd/NewCarbonDataLoadRDD.scala | 59 +++++----- .../spark/util/GlobalDictionaryUtil.scala | 5 +- .../spark/sql/catalyst/CarbonDDLSqlParser.scala | 35 +++--- .../CarbonTableIdentifierImplicit.scala | 2 +- .../spark/rdd/CarbonDataRDDFactory.scala | 1 - .../spark/sql/CarbonDatasourceRelation.scala | 10 +- .../spark/sql/CarbonDictionaryDecoder.scala | 2 +- .../scala/org/apache/spark/sql/CarbonScan.scala | 3 +- .../spark/sql/SparkUnknownExpression.scala | 9 +- .../execution/command/carbonTableSchema.scala | 9 +- .../spark/sql/optimizer/CarbonOptimizer.scala | 15 ++- .../DataCompactionBlockletBoundryTest.scala | 3 +- .../DataCompactionBoundaryConditionsTest.scala | 1 + .../DataCompactionCardinalityBoundryTest.scala | 12 +- .../datacompaction/DataCompactionLockTest.scala | 3 +- .../DataCompactionNoDictionaryTest.scala | 5 +- .../datacompaction/DataCompactionTest.scala | 12 +- .../MajorCompactionIgnoreInMinorTest.scala | 1 + .../MajorCompactionStopsAfterCompaction.scala | 1 + .../spark/util/DictionaryTestCaseUtil.scala | 3 +- .../spark/rdd/CarbonDataRDDFactory.scala | 1 - 51 files changed, 248 insertions(+), 696 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/executor/impl/DetailQueryExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/executor/impl/DetailQueryExecutor.java b/core/src/main/java/org/apache/carbondata/scan/executor/impl/DetailQueryExecutor.java index f2f4b58..7e0f5a9 100644 --- a/core/src/main/java/org/apache/carbondata/scan/executor/impl/DetailQueryExecutor.java +++ b/core/src/main/java/org/apache/carbondata/scan/executor/impl/DetailQueryExecutor.java @@ -24,6 +24,7 @@ import org.apache.carbondata.common.CarbonIterator; import org.apache.carbondata.scan.executor.exception.QueryExecutionException; import org.apache.carbondata.scan.executor.infos.BlockExecutionInfo; import org.apache.carbondata.scan.model.QueryModel; +import org.apache.carbondata.scan.result.BatchResult; import org.apache.carbondata.scan.result.iterator.DetailQueryResultIterator; /** @@ -31,13 +32,17 @@ import org.apache.carbondata.scan.result.iterator.DetailQueryResultIterator; * For executing the detail query it will pass all the block execution * info to detail query result iterator and iterator will be returned */ -public class DetailQueryExecutor extends AbstractQueryExecutor { +public class DetailQueryExecutor extends AbstractQueryExecutor<BatchResult> { - @Override public CarbonIterator<Object[]> execute(QueryModel queryModel) + @Override + public CarbonIterator<BatchResult> execute(QueryModel queryModel) throws QueryExecutionException { List<BlockExecutionInfo> blockExecutionInfoList = getBlockExecutionInfos(queryModel); - return new DetailQueryResultIterator(blockExecutionInfoList, queryModel, - queryProperties.executorService); + return new DetailQueryResultIterator( + blockExecutionInfoList, + queryModel, + queryProperties.executorService + ); } } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/executor/impl/VectorDetailQueryExecutor.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/executor/impl/VectorDetailQueryExecutor.java b/core/src/main/java/org/apache/carbondata/scan/executor/impl/VectorDetailQueryExecutor.java index 1cbf9c9..884a1eb 100644 --- a/core/src/main/java/org/apache/carbondata/scan/executor/impl/VectorDetailQueryExecutor.java +++ b/core/src/main/java/org/apache/carbondata/scan/executor/impl/VectorDetailQueryExecutor.java @@ -29,13 +29,16 @@ import org.apache.carbondata.scan.result.iterator.VectorDetailQueryResultIterato /** * Below class will be used to execute the detail query and returns columnar vectors. */ -public class VectorDetailQueryExecutor extends AbstractQueryExecutor { +public class VectorDetailQueryExecutor extends AbstractQueryExecutor<Object> { - @Override public CarbonIterator<Object[]> execute(QueryModel queryModel) + @Override public CarbonIterator<Object> execute(QueryModel queryModel) throws QueryExecutionException { List<BlockExecutionInfo> blockExecutionInfoList = getBlockExecutionInfos(queryModel); - return new VectorDetailQueryResultIterator(blockExecutionInfoList, queryModel, - queryProperties.executorService); + return new VectorDetailQueryResultIterator( + blockExecutionInfoList, + queryModel, + queryProperties.executorService + ); } } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/executor/util/QueryUtil.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/executor/util/QueryUtil.java b/core/src/main/java/org/apache/carbondata/scan/executor/util/QueryUtil.java index 3f9ebb5..9fdb9ae 100644 --- a/core/src/main/java/org/apache/carbondata/scan/executor/util/QueryUtil.java +++ b/core/src/main/java/org/apache/carbondata/scan/executor/util/QueryUtil.java @@ -336,9 +336,9 @@ public class QueryUtil { getDictionaryColumnUniqueIdentifierList(dictionaryColumnIdList, absoluteTableIdentifier.getCarbonTableIdentifier()); CacheProvider cacheProvider = CacheProvider.getInstance(); - Cache forwardDictionaryCache = cacheProvider + Cache<DictionaryColumnUniqueIdentifier, Dictionary> forwardDictionaryCache = cacheProvider .createCache(CacheType.FORWARD_DICTIONARY, absoluteTableIdentifier.getStorePath()); - List<Dictionary> columnDictionaryList = null; + List<Dictionary> columnDictionaryList; try { columnDictionaryList = forwardDictionaryCache.getAll(dictionaryColumnUniqueIdentifiers); } catch (CarbonUtilException e) { http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/filter/FilterUtil.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/filter/FilterUtil.java b/core/src/main/java/org/apache/carbondata/scan/filter/FilterUtil.java index c1985bc..943ea1a 100644 --- a/core/src/main/java/org/apache/carbondata/scan/filter/FilterUtil.java +++ b/core/src/main/java/org/apache/carbondata/scan/filter/FilterUtil.java @@ -951,12 +951,12 @@ public final class FilterUtil { new DictionaryColumnUniqueIdentifier(tableIdentifier.getCarbonTableIdentifier(), carbonDimension.getColumnIdentifier(), carbonDimension.getDataType()); CacheProvider cacheProvider = CacheProvider.getInstance(); - Cache forwardDictionaryCache = + Cache<DictionaryColumnUniqueIdentifier, Dictionary> forwardDictionaryCache = cacheProvider.createCache(CacheType.FORWARD_DICTIONARY, tableIdentifier.getStorePath()); // get the forward dictionary object - Dictionary forwardDictionary = null; + Dictionary forwardDictionary; try { - forwardDictionary = (Dictionary) forwardDictionaryCache.get(dictionaryColumnUniqueIdentifier); + forwardDictionary = forwardDictionaryCache.get(dictionaryColumnUniqueIdentifier); } catch (CarbonUtilException e) { throw new QueryExecutionException(e); } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java b/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java index ff7a93c..f6b6ba1 100644 --- a/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java +++ b/core/src/main/java/org/apache/carbondata/scan/result/iterator/AbstractDetailQueryResultIterator.java @@ -46,7 +46,7 @@ import org.apache.carbondata.scan.result.vector.CarbonColumnarBatch; * executing that query are returning a iterator over block and every time next * call will come it will execute the block and return the result */ -public abstract class AbstractDetailQueryResultIterator extends CarbonIterator { +public abstract class AbstractDetailQueryResultIterator<E> extends CarbonIterator<E> { /** * LOGGER. http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java b/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java index a958195..7d944d6 100644 --- a/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java +++ b/core/src/main/java/org/apache/carbondata/scan/result/iterator/DetailQueryResultIterator.java @@ -33,7 +33,7 @@ import org.apache.carbondata.scan.result.BatchResult; * executing that query are returning a iterator over block and every time next * call will come it will execute the block and return the result */ -public class DetailQueryResultIterator extends AbstractDetailQueryResultIterator { +public class DetailQueryResultIterator extends AbstractDetailQueryResultIterator<BatchResult> { private final Object lock = new Object(); private Future<BatchResult> future; http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/core/src/main/java/org/apache/carbondata/scan/result/iterator/VectorDetailQueryResultIterator.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/scan/result/iterator/VectorDetailQueryResultIterator.java b/core/src/main/java/org/apache/carbondata/scan/result/iterator/VectorDetailQueryResultIterator.java index 8d0e80c..417f597 100644 --- a/core/src/main/java/org/apache/carbondata/scan/result/iterator/VectorDetailQueryResultIterator.java +++ b/core/src/main/java/org/apache/carbondata/scan/result/iterator/VectorDetailQueryResultIterator.java @@ -28,7 +28,7 @@ import org.apache.carbondata.scan.result.vector.CarbonColumnarBatch; /** * It reads the data vector batch format */ -public class VectorDetailQueryResultIterator extends AbstractDetailQueryResultIterator { +public class VectorDetailQueryResultIterator extends AbstractDetailQueryResultIterator<Object> { private final Object lock = new Object(); http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedCarbonCommandException.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedCarbonCommandException.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedCarbonCommandException.java index de7d4a2..e97d063 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedCarbonCommandException.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedCarbonCommandException.java @@ -19,9 +19,6 @@ package org.apache.carbondata.spark.exception; - -import java.util.Locale; - public class MalformedCarbonCommandException extends Exception { @@ -56,17 +53,6 @@ public class MalformedCarbonCommandException extends Exception { } /** - * This method is used to get the localized message. - * - * @param locale - A Locale object represents a specific geographical, - * political, or cultural region. - * @return - Localized error message. - */ - public String getLocalizedMessage(Locale locale) { - return ""; - } - - /** * getLocalizedMessage */ @Override http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java index eca34a6..419e833 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/CarbonLoaderUtil.java @@ -241,7 +241,7 @@ public final class CarbonLoaderUtil { } } - public static void deleteStorePath(String path) { + private static void deleteStorePath(String path) { try { FileType fileType = FileFactory.getFileType(path); if (FileFactory.isFileExist(path, fileType)) { @@ -311,7 +311,7 @@ public final class CarbonLoaderUtil { String localStoreLocation = CarbonProperties.getInstance() .getProperty(tempLocationKey, CarbonCommonConstants.STORE_LOCATION_DEFAULT_VAL); try { - CarbonUtil.deleteFoldersAndFiles(new File[] { new File(localStoreLocation).getParentFile() }); + CarbonUtil.deleteFoldersAndFiles(new File(localStoreLocation).getParentFile()); LOGGER.info("Deleted the local store location" + localStoreLocation); } catch (CarbonUtilException e) { LOGGER.error(e, "Failed to delete local data load folder location"); @@ -325,15 +325,13 @@ public final class CarbonLoaderUtil { * @param storePath * @param carbonTableIdentifier * @param segmentId - * @param partitionId * @return */ public static String getStoreLocation(String storePath, - CarbonTableIdentifier carbonTableIdentifier, String segmentId, String partitionId) { + CarbonTableIdentifier carbonTableIdentifier, String segmentId) { CarbonTablePath carbonTablePath = CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier); - String carbonDataFilePath = carbonTablePath.getCarbonDataDirectoryPath(partitionId, segmentId); - return carbonDataFilePath; + return carbonTablePath.getCarbonDataDirectoryPath("0", segmentId); } /** @@ -385,9 +383,7 @@ public final class CarbonLoaderUtil { new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); if (null != listOfLoadFolderDetailsArray) { - for (LoadMetadataDetails loadMetadata : listOfLoadFolderDetailsArray) { - listOfLoadFolderDetails.add(loadMetadata); - } + Collections.addAll(listOfLoadFolderDetails, listOfLoadFolderDetailsArray); } listOfLoadFolderDetails.add(loadMetadataDetails); @@ -463,9 +459,9 @@ public final class CarbonLoaderUtil { public static Dictionary getDictionary(DictionaryColumnUniqueIdentifier columnIdentifier, String carbonStorePath) throws CarbonUtilException { - Cache dictCache = + Cache<DictionaryColumnUniqueIdentifier, Dictionary> dictCache = CacheProvider.getInstance().createCache(CacheType.REVERSE_DICTIONARY, carbonStorePath); - return (Dictionary) dictCache.get(columnIdentifier); + return dictCache.get(columnIdentifier); } public static Dictionary getDictionary(CarbonTableIdentifier tableIdentifier, http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java index 0663abc..9f092e7 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeleteLoadFolders.java @@ -26,6 +26,7 @@ * Description : for physical deletion of load folders. * Class Version : 1.0 */ + package org.apache.carbondata.spark.load; import java.io.IOException; @@ -158,59 +159,6 @@ public final class DeleteLoadFolders { return false; } - private static void factFileRenaming(String loadFolderPath) { - - FileFactory.FileType fileType = FileFactory.getFileType(loadFolderPath); - try { - if (FileFactory.isFileExist(loadFolderPath, fileType)) { - CarbonFile loadFolder = FileFactory.getCarbonFile(loadFolderPath, fileType); - - CarbonFile[] listFiles = loadFolder.listFiles(new CarbonFileFilter() { - - @Override public boolean accept(CarbonFile file) { - return (file.getName().endsWith('_' + CarbonCommonConstants.FACT_FILE_UPDATED)); - } - }); - - for (CarbonFile file : listFiles) { - if (!file.renameTo(file.getName().substring(0, - file.getName().length() - CarbonCommonConstants.FACT_FILE_UPDATED.length()))) { - LOGGER.warn("could not rename the updated fact file."); - } - } - - } - } catch (IOException e) { - LOGGER.error("exception" + e.getMessage()); - } - - } - - private static void cleanDeletedFactFile(String loadFolderPath) { - FileFactory.FileType fileType = FileFactory.getFileType(loadFolderPath); - try { - if (FileFactory.isFileExist(loadFolderPath, fileType)) { - CarbonFile loadFolder = FileFactory.getCarbonFile(loadFolderPath, fileType); - - CarbonFile[] listFiles = loadFolder.listFiles(new CarbonFileFilter() { - - @Override public boolean accept(CarbonFile file) { - return (file.getName().endsWith(CarbonCommonConstants.FACT_DELETE_EXTENSION)); - } - }); - - for (CarbonFile file : listFiles) { - if (!file.delete()) { - LOGGER.warn("could not delete the marked fact file."); - } - } - - } - } catch (IOException e) { - LOGGER.error("exception" + e.getMessage()); - } - } - public static boolean deleteLoadFoldersFromFileSystem(String dbName, String tableName, String storeLocation, boolean isForceDelete, LoadMetadataDetails[] details) { List<LoadMetadataDetails> deletedLoads = http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeletedLoadMetadata.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeletedLoadMetadata.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeletedLoadMetadata.java deleted file mode 100644 index 661e17c..0000000 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/load/DeletedLoadMetadata.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.carbondata.spark.load; - -import java.io.Serializable; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import org.apache.carbondata.core.constants.CarbonCommonConstants; - -public class DeletedLoadMetadata implements Serializable { - - private static final long serialVersionUID = 7083059404172117208L; - private Map<String, String> deletedLoadMetadataMap = - new HashMap<String, String>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); - - public void addDeletedLoadMetadata(String loadId, String status) { - deletedLoadMetadataMap.put(loadId, status); - } - - public List<String> getDeletedLoadMetadataIds() { - return new ArrayList<String>(deletedLoadMetadataMap.keySet()); - } - - public String getDeletedLoadMetadataStatus(String loadId) { - if (deletedLoadMetadataMap.containsKey(loadId)) { - return deletedLoadMetadataMap.get(loadId); - } else { - return null; - } - - } - -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonCompactionExecutor.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonCompactionExecutor.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonCompactionExecutor.java index dc5fb17..ef3d2e4 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonCompactionExecutor.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonCompactionExecutor.java @@ -56,41 +56,23 @@ public class CarbonCompactionExecutor { LogServiceFactory.getLogService(CarbonCompactionExecutor.class.getName()); private final Map<String, List<DataFileFooter>> dataFileMetadataSegMapping; private final SegmentProperties destinationSegProperties; - private final String databaseName; - private final String factTableName; private final Map<String, TaskBlockInfo> segmentMapping; - private final String storePath; private QueryExecutor queryExecutor; private CarbonTable carbonTable; private QueryModel queryModel; /** * Constructor - * - * @param segmentMapping + * @param segmentMapping * @param segmentProperties - * @param databaseName - * @param factTableName - * @param storePath * @param carbonTable */ public CarbonCompactionExecutor(Map<String, TaskBlockInfo> segmentMapping, - SegmentProperties segmentProperties, String databaseName, String factTableName, - String storePath, CarbonTable carbonTable, + SegmentProperties segmentProperties, CarbonTable carbonTable, Map<String, List<DataFileFooter>> dataFileMetadataSegMapping) { - this.segmentMapping = segmentMapping; - this.destinationSegProperties = segmentProperties; - - this.databaseName = databaseName; - - this.factTableName = factTableName; - - this.storePath = storePath; - this.carbonTable = carbonTable; - this.dataFileMetadataSegMapping = dataFileMetadataSegMapping; } @@ -143,18 +125,9 @@ public class CarbonCompactionExecutor { */ private CarbonIterator<BatchResult> executeBlockList(List<TableBlockInfo> blockList) throws QueryExecutionException { - queryModel.setTableBlockInfos(blockList); this.queryExecutor = QueryExecutorFactory.getQueryExecutor(queryModel); - CarbonIterator<BatchResult> iter = null; - try { - iter = queryExecutor.execute(queryModel); - } catch (QueryExecutionException e) { - LOGGER.error(e.getMessage()); - throw e; - } - - return iter; + return queryExecutor.execute(queryModel); } /** @@ -191,7 +164,7 @@ public class CarbonCompactionExecutor { * @param blockList * @return */ - public QueryModel prepareQueryModel(List<TableBlockInfo> blockList) { + private QueryModel prepareQueryModel(List<TableBlockInfo> blockList) { QueryModel model = new QueryModel(); http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonDataMergerUtil.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonDataMergerUtil.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonDataMergerUtil.java index 836a757..41709f7 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonDataMergerUtil.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/CarbonDataMergerUtil.java @@ -28,15 +28,12 @@ import java.util.Calendar; import java.util.Collections; import java.util.Comparator; import java.util.Date; -import java.util.HashMap; import java.util.List; -import java.util.Map; import org.apache.carbondata.common.logging.LogService; import org.apache.carbondata.common.logging.LogServiceFactory; import org.apache.carbondata.core.carbon.AbsoluteTableIdentifier; import org.apache.carbondata.core.carbon.CarbonTableIdentifier; -import org.apache.carbondata.core.carbon.datastore.block.TableBlockInfo; import org.apache.carbondata.core.carbon.path.CarbonStorePath; import org.apache.carbondata.core.carbon.path.CarbonTablePath; import org.apache.carbondata.core.constants.CarbonCommonConstants; @@ -161,7 +158,7 @@ public final class CarbonDataMergerUtil { .equalsIgnoreCase(CarbonCommonConstants.MARKED_FOR_DELETE)) { LOGGER.error("Compaction is aborted as the segment " + loadDetail.getLoadName() + " is deleted after the compaction is started."); - return tableStatusUpdationStatus; + return false; } loadDetail.setLoadStatus(CarbonCommonConstants.SEGMENT_COMPACTED); loadDetail.setModificationOrdeletionTimesStamp(modificationOrDeletionTimeStamp); @@ -225,7 +222,7 @@ public final class CarbonDataMergerUtil { CarbonLoadModel carbonLoadModel, long compactionSize, List<LoadMetadataDetails> segments, CompactionType compactionType) { - List sortedSegments = new ArrayList(segments); + List<LoadMetadataDetails> sortedSegments = new ArrayList<LoadMetadataDetails>(segments); sortSegments(sortedSegments); @@ -466,7 +463,7 @@ public final class CarbonDataMergerUtil { private static long getSizeOfSegment(String storeLocation, CarbonTableIdentifier tableIdentifier, String segId) { String loadPath = CarbonLoaderUtil - .getStoreLocation(storeLocation, tableIdentifier, segId, "0"); + .getStoreLocation(storeLocation, tableIdentifier, segId); CarbonFile segmentFolder = FileFactory.getCarbonFile(loadPath, FileFactory.getFileType(loadPath)); return getSizeOfFactFileInLoad(segmentFolder); @@ -557,11 +554,9 @@ public final class CarbonDataMergerUtil { */ private static List<LoadMetadataDetails> checkPreserveSegmentsPropertyReturnRemaining( List<LoadMetadataDetails> segments) { - - int numberOfSegmentsToBePreserved = 0; // check whether the preserving of the segments from merging is enabled or not. // get the number of loads to be preserved. - numberOfSegmentsToBePreserved = + int numberOfSegmentsToBePreserved = CarbonProperties.getInstance().getNumberOfSegmentsToBePreserved(); // get the number of valid segments and retain the latest loads from merging. return CarbonDataMergerUtil @@ -635,9 +630,9 @@ public final class CarbonDataMergerUtil { for (LoadMetadataDetails segment : loadMetadataDetails) { //check if this load is an already merged load. if (null != segment.getMergedLoadName()) { - builder.append(segment.getMergedLoadName() + ","); + builder.append(segment.getMergedLoadName()).append(","); } else { - builder.append(segment.getLoadName() + ","); + builder.append(segment.getLoadName()).append(","); } } builder.deleteCharAt(builder.length() - 1); @@ -645,37 +640,6 @@ public final class CarbonDataMergerUtil { } /** - * Combining the list of maps to one map. - * - * @param mapsOfNodeBlockMapping - * @return - */ - public static Map<String, List<TableBlockInfo>> combineNodeBlockMaps( - List<Map<String, List<TableBlockInfo>>> mapsOfNodeBlockMapping) { - - Map<String, List<TableBlockInfo>> combinedMap = - new HashMap<String, List<TableBlockInfo>>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE); - // traverse list of maps. - for (Map<String, List<TableBlockInfo>> eachMap : mapsOfNodeBlockMapping) { - // traverse inside each map. - for (Map.Entry<String, List<TableBlockInfo>> eachEntry : eachMap.entrySet()) { - - String node = eachEntry.getKey(); - List<TableBlockInfo> blocks = eachEntry.getValue(); - - // if already that node detail exist in the combined map. - if (null != combinedMap.get(node)) { - List<TableBlockInfo> blocksAlreadyPresent = combinedMap.get(node); - blocksAlreadyPresent.addAll(blocks); - } else { // if its not present in map then put to map. - combinedMap.put(node, blocks); - } - } - } - return combinedMap; - } - - /** * Removing the already merged segments from list. */ public static List<LoadMetadataDetails> filterOutNewlyAddedSegments( http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/RowResultMerger.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/RowResultMerger.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/RowResultMerger.java index 44c05a1..5c534fa 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/RowResultMerger.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/RowResultMerger.java @@ -63,7 +63,6 @@ public class RowResultMerger { private final String databaseName; private final String tableName; private final String tempStoreLocation; - private final int measureCount; private final String factStoreLocation; private CarbonFactHandler dataHandler; private List<RawResultIterator> rawResultIteratorList = @@ -101,7 +100,7 @@ public class RowResultMerger { this.databaseName = databaseName; this.tableName = tableName; - this.measureCount = segprop.getMeasures().size(); + int measureCount = segprop.getMeasures().size(); CarbonTable carbonTable = CarbonMetadata.getInstance() .getCarbonTable(databaseName + CarbonCommonConstants.UNDERSCORE + tableName); CarbonFactDataHandlerModel carbonFactDataHandlerModel = @@ -195,7 +194,7 @@ public class RowResultMerger { * * @throws SliceMergerException */ - protected void addRow(Object[] carbonTuple) throws SliceMergerException { + private void addRow(Object[] carbonTuple) throws SliceMergerException { Object[] rowInWritableFormat; rowInWritableFormat = tupleConvertor.getObjectArray(carbonTuple); @@ -270,12 +269,11 @@ public class RowResultMerger { */ private String checkAndCreateCarbonStoreLocation(String factStoreLocation, String databaseName, String tableName, String partitionId, String segmentId) { - String carbonStorePath = factStoreLocation; CarbonTable carbonTable = CarbonMetadata.getInstance() .getCarbonTable(databaseName + CarbonCommonConstants.UNDERSCORE + tableName); CarbonTableIdentifier carbonTableIdentifier = carbonTable.getCarbonTableIdentifier(); CarbonTablePath carbonTablePath = - CarbonStorePath.getCarbonTablePath(carbonStorePath, carbonTableIdentifier); + CarbonStorePath.getCarbonTablePath(factStoreLocation, carbonTableIdentifier); String carbonDataDirectoryPath = carbonTablePath.getCarbonDataDirectoryPath(partitionId, segmentId); CarbonUtil.checkAndCreateFolder(carbonDataDirectoryPath); http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/TupleConversionAdapter.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/TupleConversionAdapter.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/TupleConversionAdapter.java index 94ebf40..35b8164 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/TupleConversionAdapter.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/merger/TupleConversionAdapter.java @@ -22,19 +22,16 @@ import java.util.ArrayList; import java.util.List; import org.apache.carbondata.core.carbon.datastore.block.SegmentProperties; -import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonMeasure; import org.apache.carbondata.processing.util.RemoveDictionaryUtil; import org.apache.carbondata.scan.wrappers.ByteArrayWrapper; /** * This class will be used to convert the Result into the format used in data writer. */ -public class TupleConversionAdapter { +class TupleConversionAdapter { private final SegmentProperties segmentproperties; - private final List<CarbonMeasure> measureList; - private int noDictionaryPresentIndex; private int measureCount; @@ -48,7 +45,6 @@ public class TupleConversionAdapter { noDictionaryPresentIndex++; } this.segmentproperties = segmentProperties; - measureList = segmentProperties.getMeasures(); } /** http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/DataPartitioner.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/DataPartitioner.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/DataPartitioner.java index 58f3a2d..f097b54 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/DataPartitioner.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/DataPartitioner.java @@ -21,15 +21,7 @@ package org.apache.carbondata.spark.partition.api; import java.util.List; -import org.apache.carbondata.scan.model.CarbonQueryPlan; - -import org.apache.spark.sql.execution.command.Partitioner; - public interface DataPartitioner { - /** - * Initialise the partitioner based on the given columns - */ - void initialize(String basePath, String[] columns, Partitioner partitioner); /** * All the partitions built by the Partitioner @@ -37,18 +29,9 @@ public interface DataPartitioner { List<Partition> getAllPartitions(); /** - * Partition where the tuple should be present. (API used for data loading purpose) - */ - Partition getPartionForTuple(Object[] tuple, long rowCounter); - - /** * Identifies the partitions applicable for the given filter (API used for For query) */ - List<Partition> getPartitions(CarbonQueryPlan queryPlan); - - String[] getPartitionedColumns(); - - Partitioner getPartitioner(); + List<Partition> getPartitions(); } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/Partition.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/Partition.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/Partition.java index 61639d3..ca6e7bd 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/Partition.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/Partition.java @@ -29,11 +29,6 @@ public interface Partition extends Serializable { String getUniqueID(); /** - * File path for the raw data represented by this partition - */ - String getFilePath(); - - /** * result * * @return http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DataPartitionerProperties.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DataPartitionerProperties.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DataPartitionerProperties.java deleted file mode 100644 index bc6e54f..0000000 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DataPartitionerProperties.java +++ /dev/null @@ -1,87 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.carbondata.spark.partition.api.impl; - -import java.io.File; -import java.io.FileInputStream; -import java.io.IOException; -import java.util.Properties; - -import org.apache.carbondata.common.logging.LogService; -import org.apache.carbondata.common.logging.LogServiceFactory; - -public final class DataPartitionerProperties { - private static final LogService LOGGER = - LogServiceFactory.getLogService(DataPartitionerProperties.class.getName()); - - private static DataPartitionerProperties instance; - - private Properties properties; - - private DataPartitionerProperties() { - properties = loadProperties(); - } - - public static DataPartitionerProperties getInstance() { - if (instance == null) { - instance = new DataPartitionerProperties(); - } - return instance; - } - - public String getValue(String key, String defaultVal) { - return properties.getProperty(key, defaultVal); - } - - public String getValue(String key) { - return properties.getProperty(key); - } - - /** - * Read the properties from CSVFilePartitioner.properties - */ - private Properties loadProperties() { - Properties props = new Properties(); - - File file = new File("DataPartitioner.properties"); - FileInputStream fis = null; - try { - if (file.exists()) { - fis = new FileInputStream(file); - - props.load(fis); - } - } catch (Exception e) { - LOGGER - .error(e, e.getMessage()); - } finally { - if (null != fis) { - try { - fis.close(); - } catch (IOException e) { - LOGGER.error(e, - e.getMessage()); - } - } - } - - return props; - } -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DefaultLoadBalancer.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DefaultLoadBalancer.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DefaultLoadBalancer.java index 9bee8a2..6594c79 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DefaultLoadBalancer.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/DefaultLoadBalancer.java @@ -59,10 +59,6 @@ public class DefaultLoadBalancer { } } - public List<Partition> getPartitionsForNode(String node) { - return nodeToPartitonMap.get(node); - } - public String getNodeForPartitions(Partition partition) { return partitonToNodeMap.get(partition); } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionImpl.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionImpl.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionImpl.java deleted file mode 100644 index bd7cc42..0000000 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionImpl.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.carbondata.spark.partition.api.impl; - -import java.util.List; - -import org.apache.carbondata.spark.partition.api.Partition; - -public class PartitionImpl implements Partition { - private static final long serialVersionUID = 3020172346383028547L; - private String uniqueID; - private String folderPath; - - - public PartitionImpl(String uniqueID, String folderPath) { - this.uniqueID = uniqueID; - this.folderPath = folderPath; - } - - @Override public String getUniqueID() { - return uniqueID; - } - - @Override public String getFilePath() { - return folderPath; - } - - @Override public String toString() { - return "{PartitionID -> " + uniqueID + " Path: " + folderPath + '}'; - } - - @Override public List<String> getFilesPath() { - // TODO Auto-generated method stub - return null; - } - -} http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionMultiFileImpl.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionMultiFileImpl.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionMultiFileImpl.java index de32b5c..f34e9c9 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionMultiFileImpl.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/PartitionMultiFileImpl.java @@ -38,11 +38,6 @@ public class PartitionMultiFileImpl implements Partition { return uniqueID; } - @Override public String getFilePath() { - // TODO Auto-generated method stub - return null; - } - @Override public List<String> getFilesPath() { // TODO Auto-generated method stub return folderPath; http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/QueryPartitionHelper.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/QueryPartitionHelper.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/QueryPartitionHelper.java index e05be7d..a864ebf 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/QueryPartitionHelper.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/QueryPartitionHelper.java @@ -52,8 +52,7 @@ public final class QueryPartitionHelper { DataPartitioner dataPartitioner = partitionerMap.get(tableUniqueName); - List<Partition> queryPartitions = dataPartitioner.getPartitions(queryPlan); - return queryPartitions; + return dataPartitioner.getPartitions(); } public List<Partition> getAllPartitions(String databaseName, String tableName) { http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/SampleDataPartitionerImpl.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/SampleDataPartitionerImpl.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/SampleDataPartitionerImpl.java index c9b434a..21132ab 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/SampleDataPartitionerImpl.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/api/impl/SampleDataPartitionerImpl.java @@ -19,133 +19,24 @@ package org.apache.carbondata.spark.partition.api.impl; -import java.util.ArrayList; -import java.util.Arrays; import java.util.List; -import org.apache.carbondata.common.logging.LogService; -import org.apache.carbondata.common.logging.LogServiceFactory; -import org.apache.carbondata.core.constants.CarbonCommonConstants; -import org.apache.carbondata.scan.model.CarbonQueryPlan; import org.apache.carbondata.spark.partition.api.DataPartitioner; import org.apache.carbondata.spark.partition.api.Partition; -import org.apache.spark.sql.execution.command.Partitioner; - /** * Sample partition. */ public class SampleDataPartitionerImpl implements DataPartitioner { - private static final LogService LOGGER = - LogServiceFactory.getLogService(SampleDataPartitionerImpl.class.getName()); - private int numberOfPartitions = 1; - - private int partionColumnIndex = -1; - - private String partitionColumn; - - private Partitioner partitioner; - private List<Partition> allPartitions; - private String baseLocation; - - public SampleDataPartitionerImpl() { - } - - public void initialize(String basePath, String[] columns, Partitioner partitioner) { - this.partitioner = partitioner; - numberOfPartitions = partitioner.partitionCount(); - - partitionColumn = partitioner.partitionColumn()[0]; - LOGGER.info("SampleDataPartitionerImpl initializing with following properties."); - LOGGER.info("partitionCount: " + numberOfPartitions); - LOGGER.info("partitionColumn: " + partitionColumn); - LOGGER.info("basePath: " + basePath); - LOGGER.info("columns: " + Arrays.toString(columns)); - - this.baseLocation = basePath; - allPartitions = new ArrayList<Partition>(CarbonCommonConstants.CONSTANT_SIZE_TEN); - - for (int i = 0; i < columns.length; i++) { - if (columns[i].equalsIgnoreCase(partitionColumn)) { - partionColumnIndex = i; - break; - } - } - - for (int partionCounter = 0; partionCounter < numberOfPartitions; partionCounter++) { - PartitionImpl partitionImpl = - new PartitionImpl("" + partionCounter, baseLocation + '/' + partionCounter); - - List<Object> includedHashes = new ArrayList<Object>(CarbonCommonConstants.CONSTANT_SIZE_TEN); - includedHashes.add(partionCounter); - - allPartitions.add(partitionImpl); - } - } - - @Override public Partition getPartionForTuple(Object[] tuple, long rowCounter) { - int hashCode; - if (partionColumnIndex == -1) { - hashCode = hashCode(rowCounter); - } else { - try { - hashCode = hashCode(((String) tuple[partionColumnIndex]).hashCode()); - } catch (NumberFormatException e) { - hashCode = hashCode(0); - } - } - return allPartitions.get(hashCode); - } - /** - * - */ + @Override public List<Partition> getAllPartitions() { - return allPartitions; + return null; } - /** - * @see DataPartitioner#getPartitions(CarbonQueryPlan) - */ - public List<Partition> getPartitions(CarbonQueryPlan queryPlan) { - // TODO: this has to be redone during partitioning implmentatation - return allPartitions; - } - - /** - * Identify the partitions applicable for the given filter - */ + @Override public List<Partition> getPartitions() { - return allPartitions; - - // TODO: this has to be redone during partitioning implementation - // for (Partition aPartition : allPartitions) { - // CarbonDimensionLevelFilter partitionFilterDetails = - // aPartition.getPartitionDetails().get(partitionColumn); - // - // //Check if the partition is serving any of the - // //hash code generated for include filter of query - // for (Object includeFilter : msisdnFilter.getIncludeFilter()) { - // int hashCode = hashCode(((String) includeFilter).hashCode()); - // if (partitionFilterDetails.getIncludeFilter().contains(hashCode)) { - // allowedPartitions.add(aPartition); - // break; - // } - // } - // } - - } - - private int hashCode(long key) { - return (int) (Math.abs(key) % numberOfPartitions); - } - - @Override public String[] getPartitionedColumns() { - return new String[] { partitionColumn }; - } - - @Override public Partitioner getPartitioner() { - return partitioner; + return null; } } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/CarbonQueryUtil.java ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/CarbonQueryUtil.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/CarbonQueryUtil.java index 9d1a281..94e475e 100644 --- a/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/CarbonQueryUtil.java +++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/util/CarbonQueryUtil.java @@ -19,8 +19,8 @@ package org.apache.carbondata.spark.util; -import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -49,7 +49,7 @@ public class CarbonQueryUtil { * It creates the one split for each region server. */ public static synchronized TableSplit[] getTableSplits(String databaseName, String tableName, - CarbonQueryPlan queryPlan) throws IOException { + CarbonQueryPlan queryPlan) { //Just create splits depends on locations of region servers List<Partition> allPartitions = null; @@ -78,7 +78,7 @@ public class CarbonQueryUtil { /** * It creates the one split for each region server. */ - public static TableSplit[] getTableSplitsForDirectLoad(String sourcePath) throws Exception { + public static TableSplit[] getTableSplitsForDirectLoad(String sourcePath) { //Just create splits depends on locations of region servers DefaultLoadBalancer loadBalancer = null; @@ -104,13 +104,11 @@ public class CarbonQueryUtil { String separator) { if (StringUtils.isNotEmpty(sourcePath)) { String[] files = sourcePath.split(separator); - for (String file : files) { - partitionsFiles.add(file); - } + Collections.addAll(partitionsFiles, files); } } - private static List<Partition> getAllFilesForDataLoad(String sourcePath) throws Exception { + private static List<Partition> getAllFilesForDataLoad(String sourcePath) { List<String> files = new ArrayList<String>(CarbonCommonConstants.CONSTANT_SIZE_TEN); splitFilePath(sourcePath, files, CarbonCommonConstants.COMMA); List<Partition> partitionList = @@ -121,7 +119,7 @@ public class CarbonQueryUtil { partitionList.add(new PartitionMultiFileImpl(0 + "", partitionFiles.get(0))); for (int i = 0; i < files.size(); i++) { - partitionFiles.get(i % 1).add(files.get(i)); + partitionFiles.get(0).add(files.get(i)); } return partitionList; } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonFilters.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonFilters.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonFilters.scala index 5e58235..6115519 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonFilters.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/CarbonFilters.scala @@ -151,31 +151,30 @@ object CarbonFilters { Some(sources.EqualTo(a.name, v)) case EqualTo(Literal(v, t), Cast(a: Attribute, _)) => Some(sources.EqualTo(a.name, v)) - - case Not(EqualTo(a: Attribute, Literal(v, t))) => new - Some(sources.Not(sources.EqualTo(a.name, v))) - case Not(EqualTo(Literal(v, t), a: Attribute)) => new - Some(sources.Not(sources.EqualTo(a.name, v))) - case Not(EqualTo(Cast(a: Attribute, _), Literal(v, t))) => new - Some(sources.Not(sources.EqualTo(a.name, v))) - case Not(EqualTo(Literal(v, t), Cast(a: Attribute, _))) => new - Some(sources.Not(sources.EqualTo(a.name, v))) - case IsNotNull(a: Attribute) => Some(sources.IsNotNull(a.name)) - case IsNull(a: Attribute) => Some(sources.IsNull(a.name)) + case Not(EqualTo(a: Attribute, Literal(v, t))) => + Some(sources.Not(sources.EqualTo(a.name, v))) + case Not(EqualTo(Literal(v, t), a: Attribute)) => + Some(sources.Not(sources.EqualTo(a.name, v))) + case Not(EqualTo(Cast(a: Attribute, _), Literal(v, t))) => + Some(sources.Not(sources.EqualTo(a.name, v))) + case Not(EqualTo(Literal(v, t), Cast(a: Attribute, _))) => + Some(sources.Not(sources.EqualTo(a.name, v))) + case IsNotNull(a: Attribute) => + Some(sources.IsNotNull(a.name)) + case IsNull(a: Attribute) => + Some(sources.IsNull(a.name)) case Not(In(a: Attribute, list)) if !list.exists(!_.isInstanceOf[Literal]) => val hSet = list.map(e => e.eval(EmptyRow)) Some(sources.Not(sources.In(a.name, hSet.toArray))) case In(a: Attribute, list) if !list.exists(!_.isInstanceOf[Literal]) => val hSet = list.map(e => e.eval(EmptyRow)) Some(sources.In(a.name, hSet.toArray)) - case Not(In(Cast(a: Attribute, _), list)) - if !list.exists(!_.isInstanceOf[Literal]) => + case Not(In(Cast(a: Attribute, _), list)) if !list.exists(!_.isInstanceOf[Literal]) => val hSet = list.map(e => e.eval(EmptyRow)) Some(sources.Not(sources.In(a.name, hSet.toArray))) case In(Cast(a: Attribute, _), list) if !list.exists(!_.isInstanceOf[Literal]) => val hSet = list.map(e => e.eval(EmptyRow)) Some(sources.In(a.name, hSet.toArray)) - case GreaterThan(a: Attribute, Literal(v, t)) => Some(sources.GreaterThan(a.name, v)) case GreaterThan(Literal(v, t), a: Attribute) => @@ -184,7 +183,6 @@ object CarbonFilters { Some(sources.GreaterThan(a.name, v)) case GreaterThan(Literal(v, t), Cast(a: Attribute, _)) => Some(sources.LessThan(a.name, v)) - case LessThan(a: Attribute, Literal(v, t)) => Some(sources.LessThan(a.name, v)) case LessThan(Literal(v, t), a: Attribute) => @@ -193,7 +191,6 @@ object CarbonFilters { Some(sources.LessThan(a.name, v)) case LessThan(Literal(v, t), Cast(a: Attribute, _)) => Some(sources.GreaterThan(a.name, v)) - case GreaterThanOrEqual(a: Attribute, Literal(v, t)) => Some(sources.GreaterThanOrEqual(a.name, v)) case GreaterThanOrEqual(Literal(v, t), a: Attribute) => @@ -202,7 +199,6 @@ object CarbonFilters { Some(sources.GreaterThanOrEqual(a.name, v)) case GreaterThanOrEqual(Literal(v, t), Cast(a: Attribute, _)) => Some(sources.LessThanOrEqual(a.name, v)) - case LessThanOrEqual(a: Attribute, Literal(v, t)) => Some(sources.LessThanOrEqual(a.name, v)) case LessThanOrEqual(Literal(v, t), a: Attribute) => @@ -248,23 +244,63 @@ object CarbonFilters { (transformExpression(left) ++ transformExpression(right)).reduceOption(new AndExpression(_, _)) - case EqualTo(a: Attribute, l@Literal(v, t)) => new - Some(new EqualToExpression(transformExpression(a).get, transformExpression(l).get)) - case EqualTo(l@Literal(v, t), a: Attribute) => new - Some(new EqualToExpression(transformExpression(a).get, transformExpression(l).get)) - case EqualTo(Cast(a: Attribute, _), l@Literal(v, t)) => new - Some(new EqualToExpression(transformExpression(a).get, transformExpression(l).get)) - case EqualTo(l@Literal(v, t), Cast(a: Attribute, _)) => new - Some(new EqualToExpression(transformExpression(a).get, transformExpression(l).get)) - - case Not(EqualTo(a: Attribute, l@Literal(v, t))) => new - Some(new NotEqualsExpression(transformExpression(a).get, transformExpression(l).get)) - case Not(EqualTo(l@Literal(v, t), a: Attribute)) => new - Some(new NotEqualsExpression(transformExpression(a).get, transformExpression(l).get)) - case Not(EqualTo(Cast(a: Attribute, _), l@Literal(v, t))) => new - Some(new NotEqualsExpression(transformExpression(a).get, transformExpression(l).get)) - case Not(EqualTo(l@Literal(v, t), Cast(a: Attribute, _))) => new - Some(new NotEqualsExpression(transformExpression(a).get, transformExpression(l).get)) + case EqualTo(a: Attribute, l@Literal(v, t)) => + Some( + new EqualToExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + case EqualTo(l@Literal(v, t), a: Attribute) => + Some( + new EqualToExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + case EqualTo(Cast(a: Attribute, _), l@Literal(v, t)) => + Some( + new EqualToExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + case EqualTo(l@Literal(v, t), Cast(a: Attribute, _)) => + Some( + new EqualToExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + + case Not(EqualTo(a: Attribute, l@Literal(v, t))) => + Some( + new NotEqualsExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + case Not(EqualTo(l@Literal(v, t), a: Attribute)) => + Some( + new NotEqualsExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + case Not(EqualTo(Cast(a: Attribute, _), l@Literal(v, t))) => + Some( + new NotEqualsExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) + case Not(EqualTo(l@Literal(v, t), Cast(a: Attribute, _))) => + Some( + new NotEqualsExpression( + transformExpression(a).get, + transformExpression(l).get + ) + ) case IsNotNull(child: Attribute) => Some(new NotEqualsExpression(transformExpression(child).get, transformExpression(Literal(null)).get, true)) http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala index 5d6a663..57087f3 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataLoadRDD.scala @@ -231,8 +231,8 @@ class DataFileLoaderRDD[K, V]( var partitionID = "0" val loadMetadataDetails = new LoadMetadataDetails() var model: CarbonLoadModel = _ - var uniqueLoadStatusId = carbonLoadModel.getTableName + CarbonCommonConstants.UNDERSCORE + - theSplit.index + val uniqueLoadStatusId = carbonLoadModel.getTableName + CarbonCommonConstants.UNDERSCORE + + theSplit.index try { loadMetadataDetails.setPartitionCount(partitionID) carbonLoadModel.setSegmentId(String.valueOf(loadCount)) @@ -500,10 +500,10 @@ class DataFrameLoaderRDD[K, V]( override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = { val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) val resultIter = new Iterator[(K, V)] { - var partitionID = "0" + val partitionID = "0" val loadMetadataDetails = new LoadMetadataDetails() - var uniqueLoadStatusId = carbonLoadModel.getTableName + CarbonCommonConstants.UNDERSCORE + - theSplit.index + val uniqueLoadStatusId = carbonLoadModel.getTableName + CarbonCommonConstants.UNDERSCORE + + theSplit.index try { loadMetadataDetails.setPartitionCount(partitionID) carbonLoadModel.setPartitionId(partitionID) http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDeleteLoadByDateRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDeleteLoadByDateRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDeleteLoadByDateRDD.scala index 0534def..33f453b 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDeleteLoadByDateRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDeleteLoadByDateRDD.scala @@ -25,7 +25,6 @@ import org.apache.spark.rdd.RDD import org.apache.carbondata.core.constants.CarbonCommonConstants import org.apache.carbondata.core.load.LoadMetadataDetails import org.apache.carbondata.spark.DeletedLoadResult -import org.apache.carbondata.spark.load.DeletedLoadMetadata import org.apache.carbondata.spark.util.CarbonQueryUtil class CarbonDeleteLoadByDateRDD[K, V]( @@ -53,7 +52,6 @@ class CarbonDeleteLoadByDateRDD[K, V]( override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = { new Iterator[(K, V)] { - val deletedMetaData = new DeletedLoadMetadata() val split = theSplit.asInstanceOf[CarbonLoadPartition] logInfo("Input split: " + split.serializableHadoopSplit.value) @@ -69,7 +67,7 @@ class CarbonDeleteLoadByDateRDD[K, V]( case e: Exception => logInfo("Unable to parse with default time format " + dateValue) } // TODO: Implement it - var finished = false + val finished = false override def hasNext: Boolean = { finished http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala index 3c15818..9d9fa99 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonGlobalDictionaryRDD.scala @@ -23,7 +23,7 @@ import java.text.SimpleDateFormat import java.util.regex.Pattern import scala.collection.mutable -import scala.collection.mutable.{ArrayBuffer, HashMap, HashSet} +import scala.collection.mutable.ArrayBuffer import scala.util.control.Breaks.{break, breakable} import au.com.bytecode.opencsv.CSVReader @@ -34,17 +34,19 @@ import org.apache.spark.sql.Row import org.apache.carbondata.common.factory.CarbonCommonFactory import org.apache.carbondata.common.logging.LogServiceFactory +import org.apache.carbondata.core.cache.dictionary.Dictionary import org.apache.carbondata.core.carbon.{CarbonTableIdentifier, ColumnIdentifier} import org.apache.carbondata.core.carbon.metadata.schema.table.column.CarbonDimension +import org.apache.carbondata.core.carbon.path.CarbonTablePath import org.apache.carbondata.core.constants.CarbonCommonConstants import org.apache.carbondata.core.datastorage.store.impl.FileFactory +import org.apache.carbondata.core.service.PathService import org.apache.carbondata.core.util.{CarbonProperties, CarbonTimeStatisticsFactory, CarbonUtil} import org.apache.carbondata.lcm.locks.{CarbonLockFactory, LockUsage} import org.apache.carbondata.processing.model.CarbonLoadModel import org.apache.carbondata.spark.load.CarbonLoaderUtil import org.apache.carbondata.spark.tasks.{DictionaryWriterTask, SortIndexWriterTask} -import org.apache.carbondata.spark.util.CarbonScalaUtil -import org.apache.carbondata.spark.util.GlobalDictionaryUtil +import org.apache.carbondata.spark.util.{CarbonScalaUtil, GlobalDictionaryUtil} /** * A partitioner partition by column. @@ -70,9 +72,9 @@ case class DictionaryStats(distinctValues: java.util.List[String], dictWriteTime: Long, sortIndexWriteTime: Long) case class PrimitiveParser(dimension: CarbonDimension, - setOpt: Option[HashSet[String]]) extends GenericParser { - val (hasDictEncoding, set: HashSet[String]) = setOpt match { - case None => (false, new HashSet[String]) + setOpt: Option[mutable.HashSet[String]]) extends GenericParser { + val (hasDictEncoding, set: mutable.HashSet[String]) = setOpt match { + case None => (false, new mutable.HashSet[String]) case Some(x) => (true, x) } @@ -183,7 +185,7 @@ class CarbonAllDictionaryCombineRDD( ): Iterator[(Int, ColumnDistinctValues)] = { val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) - val distinctValuesList = new ArrayBuffer[(Int, HashSet[String])] + val distinctValuesList = new ArrayBuffer[(Int, mutable.HashSet[String])] /* * for all dictionary, all columns need to encoding and checking * isHighCardinalityColumn, so no need to calculate rowcount @@ -194,7 +196,7 @@ class CarbonAllDictionaryCombineRDD( GlobalDictionaryUtil.createDimensionParsers(model, distinctValuesList) val dimNum = model.dimensions.length // Map[dimColName -> dimColNameIndex] - val columnIndexMap = new HashMap[String, Int]() + val columnIndexMap = new mutable.HashMap[String, Int]() for (j <- 0 until dimNum) { columnIndexMap.put(model.dimensions(j).getColName, j) } @@ -245,7 +247,7 @@ class CarbonBlockDistinctValuesCombineRDD( context: TaskContext): Iterator[(Int, ColumnDistinctValues)] = { val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) CarbonTimeStatisticsFactory.getLoadStatisticsInstance.recordLoadCsvfilesToDfTime() - val distinctValuesList = new ArrayBuffer[(Int, HashSet[String])] + val distinctValuesList = new ArrayBuffer[(Int, mutable.HashSet[String])] var rowCount = 0L try { val dimensionParsers = @@ -297,15 +299,15 @@ class CarbonGlobalDictionaryGenerateRDD( override def compute(split: Partition, context: TaskContext): Iterator[(Int, String, Boolean)] = { val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) - var status = CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS + val status = CarbonCommonConstants.STORE_LOADSTATUS_SUCCESS var isHighCardinalityColumn = false val iter = new Iterator[(Int, String, Boolean)] { - var dictionaryForDistinctValueLookUp: - org.apache.carbondata.core.cache.dictionary.Dictionary = _ - var dictionaryForSortIndexWriting: org.apache.carbondata.core.cache.dictionary.Dictionary = _ + var dictionaryForDistinctValueLookUp: Dictionary = _ + var dictionaryForSortIndexWriting: Dictionary = _ var dictionaryForDistinctValueLookUpCleared: Boolean = false - val pathService = CarbonCommonFactory.getPathService - val carbonTablePath = pathService.getCarbonTablePath(model.hdfsLocation, model.table) + val pathService: PathService = CarbonCommonFactory.getPathService + val carbonTablePath: CarbonTablePath = + pathService.getCarbonTablePath(model.hdfsLocation, model.table) if (StringUtils.isNotBlank(model.hdfsTempLocation )) { CarbonProperties.getInstance.addProperty(CarbonCommonConstants.HDFS_TEMP_LOCATION, model.hdfsTempLocation) @@ -538,8 +540,8 @@ class CarbonColumnDictGenerateRDD(carbonLoadModel: CarbonLoadModel, } } } - val mapIdWithSet = new HashMap[String, HashSet[String]] - val columnValues = new HashSet[String] + val mapIdWithSet = new mutable.HashMap[String, mutable.HashSet[String]] + val columnValues = new mutable.HashSet[String] val distinctValues = (theSplit.index, columnValues) mapIdWithSet.put(primDimension.getColumnId, columnValues) // use parser to generate new dict value http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala index e740caa..94fa601 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala @@ -123,10 +123,8 @@ class CarbonMergerRDD[K, V]( carbonLoadModel.setStorePath(storePath) - exec = new CarbonCompactionExecutor(segmentMapping, segmentProperties, databaseName, - factTableName, storePath, carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable, - dataFileMetadataSegMapping - ) + exec = new CarbonCompactionExecutor(segmentMapping, segmentProperties, + carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable, dataFileMetadataSegMapping) // fire a query and get the results. var result2: util.List[RawResultIterator] = null http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataLoadPartitionCoalescer.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataLoadPartitionCoalescer.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataLoadPartitionCoalescer.scala index 77402b4..02718e0 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataLoadPartitionCoalescer.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataLoadPartitionCoalescer.scala @@ -71,7 +71,7 @@ class DataLoadPartitionCoalescer(prev: RDD[_], nodeList: Array[String]) { private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName) val prevPartitions = prev.partitions - var numOfParts = Math.max(1, Math.min(nodeList.length, prevPartitions.length)) + val numOfParts = Math.max(1, Math.min(nodeList.length, prevPartitions.length)) // host => partition id list val hostMapPartitionIds = new HashMap[String, LinkedHashSet[Int]] // partition id => host list @@ -268,7 +268,7 @@ class DataLoadPartitionCoalescer(prev: RDD[_], nodeList: Array[String]) { prevPartIndexs(i % numOfParts) += prevPartitions(i).index } prevPartIndexs.filter(_.nonEmpty).zipWithIndex.map { x => - new CoalescedRDDPartition(x._2, prev, x._1.toArray, getLocation(x._2)) + CoalescedRDDPartition(x._2, prev, x._1.toArray, getLocation(x._2)) } } @@ -312,8 +312,8 @@ class DataLoadPartitionCoalescer(prev: RDD[_], nodeList: Array[String]) { } else { Some(emptyHosts(index - localityResult.length)) } - LOGGER.info(s"CoalescedRDDPartition ${index}, ${ids.length}, ${loc} ") - new CoalescedRDDPartition(index, prev, ids, loc) + LOGGER.info(s"CoalescedRDDPartition $index, ${ids.length}, $loc ") + CoalescedRDDPartition(index, prev, ids, loc) }.filter(_.parentsIndices.nonEmpty).toArray } http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala index 64b8b61..22d8406 100644 --- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala +++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/NewCarbonDataLoadRDD.scala @@ -144,7 +144,7 @@ class NewCarbonDataLoadRDD[K, V]( var partitionID = "0" val loadMetadataDetails = new LoadMetadataDetails() var model: CarbonLoadModel = _ - var uniqueLoadStatusId = + val uniqueLoadStatusId = carbonLoadModel.getTableName + CarbonCommonConstants.UNDERSCORE + theSplit.index try { loadMetadataDetails.setPartitionCount(partitionID) @@ -279,38 +279,35 @@ class NewCarbonDataLoadRDD[K, V]( } override def getPreferredLocations(split: Partition): Seq[String] = { - isTableSplitPartition match { - case true => - // for table split partition - val theSplit = split.asInstanceOf[CarbonTableSplitPartition] - val location = theSplit.serializableHadoopSplit.value.getLocations.asScala - location - case false => - // for node partition - val theSplit = split.asInstanceOf[CarbonNodePartition] - val firstOptionLocation: Seq[String] = List(theSplit.serializableHadoopSplit) - logInfo("Preferred Location for split : " + firstOptionLocation.head) - val blockMap = new util.LinkedHashMap[String, Integer]() - val tableBlocks = theSplit.blocksDetails - tableBlocks.foreach { tableBlock => - tableBlock.getLocations.foreach { location => - if (!firstOptionLocation.exists(location.equalsIgnoreCase(_))) { - val currentCount = blockMap.get(location) - if (currentCount == null) { - blockMap.put(location, 1) - } else { - blockMap.put(location, currentCount + 1) - } + if (isTableSplitPartition) { + val theSplit = split.asInstanceOf[CarbonTableSplitPartition] + val location = theSplit.serializableHadoopSplit.value.getLocations.asScala + location + } else { + val theSplit = split.asInstanceOf[CarbonNodePartition] + val firstOptionLocation: Seq[String] = List(theSplit.serializableHadoopSplit) + logInfo("Preferred Location for split : " + firstOptionLocation.head) + val blockMap = new util.LinkedHashMap[String, Integer]() + val tableBlocks = theSplit.blocksDetails + tableBlocks.foreach { tableBlock => + tableBlock.getLocations.foreach { location => + if (!firstOptionLocation.exists(location.equalsIgnoreCase(_))) { + val currentCount = blockMap.get(location) + if (currentCount == null) { + blockMap.put(location, 1) + } else { + blockMap.put(location, currentCount + 1) } } } + } - val sortedList = blockMap.entrySet().asScala.toSeq.sortWith {(nodeCount1, nodeCount2) => - nodeCount1.getValue > nodeCount2.getValue - } + val sortedList = blockMap.entrySet().asScala.toSeq.sortWith { (nodeCount1, nodeCount2) => + nodeCount1.getValue > nodeCount2.getValue + } - val sortedNodesList = sortedList.map(nodeCount => nodeCount.getKey).take(2) - firstOptionLocation ++ sortedNodesList + val sortedNodesList = sortedList.map(nodeCount => nodeCount.getKey).take(2) + firstOptionLocation ++ sortedNodesList } } } @@ -333,10 +330,10 @@ class NewDataFrameLoaderRDD[K, V]( override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = { val LOGGER = LogServiceFactory.getLogService(this.getClass.getName) val iter = new Iterator[(K, V)] { - var partitionID = "0" + val partitionID = "0" val loadMetadataDetails = new LoadMetadataDetails() - var model: CarbonLoadModel = carbonLoadModel - var uniqueLoadStatusId = + val model: CarbonLoadModel = carbonLoadModel + val uniqueLoadStatusId = carbonLoadModel.getTableName + CarbonCommonConstants.UNDERSCORE + theSplit.index try { http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/7b8b1959/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 5d6badd..84c71ed 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 @@ -328,7 +328,7 @@ object GlobalDictionaryUtil { if (null == carbonLoadModel.getLoadMetadataDetails) { CommonUtil.readLoadMetadataDetails(carbonLoadModel, hdfsLocation) } - new DictionaryLoadModel(table, + DictionaryLoadModel(table, dimensions, hdfsLocation, dictfolderPath, @@ -345,7 +345,8 @@ object GlobalDictionaryUtil { hdfsTempLocation, lockType, zookeeperUrl, - serializationNullFormat) + serializationNullFormat + ) } /**
