You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2016/11/19 02:16:51 UTC
[2/4] incubator-carbondata git commit: Improved spark module code. *
Removed some compliation warnings. * Replace pattern matching for boolean to
IF-ELSE. * Improved code according to scala standards. * Removed unnecessary
new lines. * Added string inter
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
index b8f4087..1a7aaf6 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonMergerRDD.scala
@@ -32,13 +32,15 @@ import org.apache.spark.sql.hive.DistributionUtil
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.carbon.{AbsoluteTableIdentifier, CarbonTableIdentifier}
-import org.apache.carbondata.core.carbon.datastore.block.{Distributable, SegmentProperties, TableBlockInfo, TableTaskInfo, TaskBlockInfo}
+import org.apache.carbondata.core.carbon.datastore.block.{Distributable, SegmentProperties,
+TableBlockInfo, TableTaskInfo, TaskBlockInfo}
import org.apache.carbondata.core.carbon.metadata.blocklet.DataFileFooter
import org.apache.carbondata.core.carbon.path.CarbonTablePath
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil, CarbonUtilException}
import org.apache.carbondata.hadoop.{CarbonInputFormat, CarbonInputSplit}
-import org.apache.carbondata.integration.spark.merger.{CarbonCompactionExecutor, CarbonCompactionUtil, RowResultMerger}
+import org.apache.carbondata.integration.spark.merger.{CarbonCompactionExecutor,
+CarbonCompactionUtil, RowResultMerger}
import org.apache.carbondata.processing.model.CarbonLoadModel
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil
import org.apache.carbondata.scan.result.iterator.RawResultIterator
@@ -49,11 +51,11 @@ import org.apache.carbondata.spark.util.QueryPlanUtil
class CarbonMergerRDD[K, V](
- sc: SparkContext,
- result: MergeResult[K, V],
- carbonLoadModel: CarbonLoadModel,
- carbonMergerMapping : CarbonMergerMapping,
- confExecutorsTemp: String)
+ sc: SparkContext,
+ result: MergeResult[K, V],
+ carbonLoadModel: CarbonLoadModel,
+ carbonMergerMapping: CarbonMergerMapping,
+ confExecutorsTemp: String)
extends RDD[(K, V)](sc, Nil) with Logging {
sc.setLocalProperty("spark.scheduler.pool", "DDL")
@@ -66,37 +68,37 @@ class CarbonMergerRDD[K, V](
val databaseName = carbonMergerMapping.databaseName
val factTableName = carbonMergerMapping.factTableName
val tableId = carbonMergerMapping.tableId
+
override def compute(theSplit: Partition, context: TaskContext): Iterator[(K, V)] = {
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
val iter = new Iterator[(K, V)] {
carbonLoadModel.setTaskNo(String.valueOf(theSplit.index))
val tempLocationKey: String = CarbonCommonConstants
- .COMPACTION_KEY_WORD + '_' + carbonLoadModel
- .getDatabaseName + '_' + carbonLoadModel
- .getTableName + '_' + carbonLoadModel.getTaskNo
+ .COMPACTION_KEY_WORD + '_' + carbonLoadModel
+ .getDatabaseName + '_' + carbonLoadModel
+ .getTableName + '_' + carbonLoadModel.getTaskNo
// this property is used to determine whether temp location for carbon is inside
// container temp dir or is yarn application directory.
val carbonUseLocalDir = CarbonProperties.getInstance()
.getProperty("carbon.use.local.dir", "false")
- if(carbonUseLocalDir.equalsIgnoreCase("true")) {
+ if (carbonUseLocalDir.equalsIgnoreCase("true")) {
val storeLocations = CarbonLoaderUtil.getConfiguredLocalDirs(SparkEnv.get.conf)
- if (null != storeLocations && storeLocations.length > 0) {
+ if (null != storeLocations && storeLocations.nonEmpty) {
storeLocation = storeLocations(Random.nextInt(storeLocations.length))
}
if (storeLocation == null) {
storeLocation = System.getProperty("java.io.tmpdir")
}
- }
- else {
+ } else {
storeLocation = System.getProperty("java.io.tmpdir")
}
storeLocation = storeLocation + '/' + System.nanoTime() + '/' + theSplit.index
CarbonProperties.getInstance().addProperty(tempLocationKey, storeLocation)
- LOGGER.info("Temp storeLocation taken is " + storeLocation)
+ LOGGER.info(s"Temp storeLocation taken is $storeLocation")
var mergeStatus = false
var mergeNumber = ""
var exec: CarbonCompactionExecutor = null
@@ -111,7 +113,7 @@ class CarbonMergerRDD[K, V](
carbonMergerMapping.maxSegmentColCardinality)
// sorting the table block info List.
- var tableBlockInfoList = carbonSparkPartition.tableBlockInfos
+ val tableBlockInfoList = carbonSparkPartition.tableBlockInfos
Collections.sort(tableBlockInfoList)
@@ -123,7 +125,7 @@ class CarbonMergerRDD[K, V](
carbonLoadModel.setStorePath(storePath)
- exec = new CarbonCompactionExecutor(segmentMapping, segmentProperties, databaseName,
+ exec = new CarbonCompactionExecutor(segmentMapping, segmentProperties, databaseName,
factTableName, storePath, carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable,
dataFileMetadataSegMapping
)
@@ -135,18 +137,18 @@ class CarbonMergerRDD[K, V](
} catch {
case e: Throwable =>
if (null != exec) {
- exec.finish
+ exec.finish()
}
LOGGER.error(e)
if (null != e.getMessage) {
- sys.error("Exception occurred in query execution :: " + e.getMessage)
+ sys.error(s"Exception occurred in query execution :: ${ e.getMessage }")
} else {
sys.error("Exception occurred in query execution.Please check logs.")
}
}
mergeNumber = mergedLoadName
.substring(mergedLoadName.lastIndexOf(CarbonCommonConstants.LOAD_FOLDER) +
- CarbonCommonConstants.LOAD_FOLDER.length(), mergedLoadName.length()
+ CarbonCommonConstants.LOAD_FOLDER.length(), mergedLoadName.length()
)
val tempStoreLoc = CarbonDataProcessorUtil.getLocalDataFolderLocation(databaseName,
@@ -170,13 +172,11 @@ class CarbonMergerRDD[K, V](
)
mergeStatus = merger.mergerSlice()
- }
- catch {
+ } catch {
case e: Exception =>
LOGGER.error(e)
throw e
- }
- finally {
+ } finally {
// delete temp location data
val newSlice = CarbonCommonConstants.LOAD_FOLDER + mergeNumber
try {
@@ -187,9 +187,9 @@ class CarbonMergerRDD[K, V](
case e: Exception =>
LOGGER.error(e)
}
- if (null != exec) {
- exec.finish
- }
+ if (null != exec) {
+ exec.finish
+ }
}
var finished = false
@@ -198,8 +198,7 @@ class CarbonMergerRDD[K, V](
if (!finished) {
finished = true
finished
- }
- else {
+ } else {
!finished
}
}
@@ -261,7 +260,7 @@ class CarbonMergerRDD[K, V](
)
// keep on assigning till last one is reached.
- if (null != blocksOfOneSegment && blocksOfOneSegment.size > 0) {
+ if (null != blocksOfOneSegment && blocksOfOneSegment.nonEmpty) {
blocksOfLastSegment = blocksOfOneSegment.asJava
}
@@ -273,8 +272,7 @@ class CarbonMergerRDD[K, V](
val blockListTemp = new util.ArrayList[TableBlockInfo]()
blockListTemp.add(tableBlockInfo)
taskIdMapping.put(taskNo, blockListTemp)
- }
- else {
+ } else {
blockList.add(tableBlockInfo)
}
}
@@ -288,8 +286,7 @@ class CarbonMergerRDD[K, V](
}
// prepare the details required to extract the segment properties using last segment.
- if (null != blocksOfLastSegment && blocksOfLastSegment.size > 0)
- {
+ if (null != blocksOfLastSegment && blocksOfLastSegment.size > 0) {
val lastBlockInfo = blocksOfLastSegment.get(blocksOfLastSegment.size - 1)
var dataFileFooter: DataFileFooter = null
@@ -317,8 +314,8 @@ class CarbonMergerRDD[K, V](
} else { nodeMapping.size() }
CarbonContext.ensureExecutors(sparkContext, requiredExecutors)
logInfo("No.of Executors required=" + requiredExecutors
- + " , spark.executor.instances=" + confExecutors
- + ", no.of.nodes where data present=" + nodeMapping.size())
+ + " , spark.executor.instances=" + confExecutors
+ + ", no.of.nodes where data present=" + nodeMapping.size())
var nodes = DistributionUtil.getNodeList(sparkContext)
var maxTimes = 30
while (nodes.length < requiredExecutors && maxTimes > 0) {
@@ -341,11 +338,11 @@ class CarbonMergerRDD[K, V](
val list = new util.ArrayList[TableBlockInfo]
entry._2.asScala.foreach(taskInfo => {
- val blocksPerNode = taskInfo.asInstanceOf[TableTaskInfo]
- list.addAll(blocksPerNode.getTableBlockInfoList)
+ val blocksPerNode = taskInfo.asInstanceOf[TableTaskInfo]
+ list.addAll(blocksPerNode.getTableBlockInfoList)
taskBlockList
.add(new NodeInfo(blocksPerNode.getTaskId, blocksPerNode.getTableBlockInfoList.size))
- })
+ })
if (list.size() != 0) {
result.add(new CarbonSparkPartition(id, i, Seq(entry._1).toArray, list))
i += 1
@@ -354,25 +351,25 @@ class CarbonMergerRDD[K, V](
// print the node info along with task and number of blocks for the task.
- nodeTaskBlocksMap.asScala.foreach((entry : (String, List[NodeInfo])) => {
- logInfo(s"for the node ${entry._1}" )
+ nodeTaskBlocksMap.asScala.foreach((entry: (String, List[NodeInfo])) => {
+ logInfo(s"for the node ${ entry._1 }")
for (elem <- entry._2.asScala) {
logInfo("Task ID is " + elem.TaskId + "no. of blocks is " + elem.noOfBlocks)
}
- } )
+ })
val noOfNodes = nodes.length
val noOfTasks = result.size
logInfo(s"Identified no.of.Blocks: $noOfBlocks,"
+ s"parallelism: $defaultParallelism , no.of.nodes: $noOfNodes, no.of.tasks: $noOfTasks"
)
- logInfo("Time taken to identify Blocks to scan : " + (System
- .currentTimeMillis() - startTime)
+ logInfo("Time taken to identify Blocks to scan: " + (System
+ .currentTimeMillis() - startTime)
)
- for (j <- 0 until result.size ) {
+ for (j <- 0 until result.size) {
val cp = result.get(j).asInstanceOf[CarbonSparkPartition]
- logInfo(s"Node : " + cp.locations.toSeq.mkString(",")
- + ", No.Of Blocks : " + cp.tableBlockInfos.size
+ logInfo(s"Node: " + cp.locations.toSeq.mkString(",")
+ + ", No.Of Blocks: " + cp.tableBlockInfos.size
)
}
result.toArray(new Array[Partition](result.size))
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index c496099..d56b00f 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -49,7 +49,6 @@ import org.apache.carbondata.spark.load.CarbonLoaderUtil
import org.apache.carbondata.spark.util.QueryPlanUtil
-
class CarbonSparkPartition(rddId: Int, val idx: Int,
val locations: Array[String],
val tableBlockInfos: util.List[TableBlockInfo])
@@ -92,7 +91,7 @@ class CarbonScanRDD[V: ClassTag](
val result = new util.ArrayList[Partition](defaultParallelism)
val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
val validAndInvalidSegments = new SegmentStatusManager(queryModel.getAbsoluteTableIdentifier)
- .getValidAndInvalidSegments
+ .getValidAndInvalidSegments
// set filter resolver tree
try {
// before applying filter check whether segments are available in the table.
@@ -110,11 +109,10 @@ class CarbonScanRDD[V: ClassTag](
queryModel.getAbsoluteTableIdentifier
)
}
- }
- catch {
+ } catch {
case e: Exception =>
LOGGER.error(e)
- sys.error("Exception occurred in query execution :: " + e.getMessage)
+ sys.error(s"Exception occurred in query execution :: ${ e.getMessage }")
}
// get splits
val splits = carbonInputFormat.getSplits(job)
@@ -129,8 +127,8 @@ class CarbonScanRDD[V: ClassTag](
)
)
var activeNodes = Array[String]()
- if(blockListTemp.nonEmpty) {
- activeNodes = DistributionUtil
+ if (blockListTemp.nonEmpty) {
+ activeNodes = DistributionUtil
.ensureExecutorsAndGetNodeList(blockListTemp.toArray, sparkContext)
}
defaultParallelism = sparkContext.defaultParallelism
@@ -141,9 +139,9 @@ class CarbonScanRDD[V: ClassTag](
var statistic = new QueryStatistic()
// group blocks to nodes, tasks
val nodeBlockMapping =
- CarbonLoaderUtil.nodeBlockTaskMapping(blockList.asJava, -1, defaultParallelism,
- activeNodes.toList.asJava
- )
+ CarbonLoaderUtil.nodeBlockTaskMapping(blockList.asJava, -1, defaultParallelism,
+ activeNodes.toList.asJava
+ )
statistic.addStatistics(QueryStatisticsConstants.BLOCK_ALLOCATION, System.currentTimeMillis)
statisticRecorder.recordStatisticsForDriver(statistic, queryModel.getQueryId())
statistic = new QueryStatistic()
@@ -173,15 +171,14 @@ class CarbonScanRDD[V: ClassTag](
statisticRecorder.logStatisticsAsTableDriver()
result.asScala.foreach { r =>
val cp = r.asInstanceOf[CarbonSparkPartition]
- logInfo(s"Node : " + cp.locations.toSeq.mkString(",")
- + ", No.Of Blocks : " + cp.tableBlockInfos.size()
+ logInfo(s"Node: ${ cp.locations.toSeq.mkString(",") }" +
+ s", No.Of Blocks: ${ cp.tableBlockInfos.size() }"
)
}
} else {
logInfo("No blocks identified to scan")
}
- }
- else {
+ } else {
logInfo("No valid segments found to scan")
}
result.toArray(new Array[Partition](result.size()))
@@ -200,7 +197,7 @@ class CarbonScanRDD[V: ClassTag](
queryExecutor.finish
})
val carbonSparkPartition = thepartition.asInstanceOf[CarbonSparkPartition]
- if(!carbonSparkPartition.tableBlockInfos.isEmpty) {
+ if (!carbonSparkPartition.tableBlockInfos.isEmpty) {
queryModel.setQueryId(queryModel.getQueryId + "_" + carbonSparkPartition.idx)
// fill table block info
queryModel.setTableBlockInfos(carbonSparkPartition.tableBlockInfos)
@@ -221,7 +218,7 @@ class CarbonScanRDD[V: ClassTag](
case e: Exception =>
LOGGER.error(e)
if (null != e.getMessage) {
- sys.error("Exception occurred in query execution :: " + e.getMessage)
+ sys.error(s"Exception occurred in query execution :: ${ e.getMessage }")
} else {
sys.error("Exception occurred in query execution.Please check logs.")
}
@@ -256,22 +253,22 @@ class CarbonScanRDD[V: ClassTag](
}
def logStatistics(): Unit = {
- if (null != queryModel.getStatisticsRecorder) {
- var queryStatistic = new QueryStatistic()
- queryStatistic
- .addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,
- System.currentTimeMillis - queryStartTime
- )
- queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
- // result size
- queryStatistic = new QueryStatistic()
- queryStatistic.addCountStatistic(QueryStatisticsConstants.RESULT_SIZE, recordCount)
- queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
- // print executor query statistics for each task_id
- queryModel.getStatisticsRecorder.logStatisticsAsTableExecutor()
- }
+ if (null != queryModel.getStatisticsRecorder) {
+ var queryStatistic = new QueryStatistic()
+ queryStatistic
+ .addFixedTimeStatistic(QueryStatisticsConstants.EXECUTOR_PART,
+ System.currentTimeMillis - queryStartTime
+ )
+ queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
+ // result size
+ queryStatistic = new QueryStatistic()
+ queryStatistic.addCountStatistic(QueryStatisticsConstants.RESULT_SIZE, recordCount)
+ queryModel.getStatisticsRecorder.recordStatistics(queryStatistic)
+ // print executor query statistics for each task_id
+ queryModel.getStatisticsRecorder.logStatisticsAsTableExecutor()
}
}
+ }
iter
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
index 28c37f3..9c9be8d 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/rdd/Compactor.scala
@@ -50,18 +50,18 @@ object Compactor {
val carbonLoadModel = compactionCallableModel.carbonLoadModel
val compactionType = compactionCallableModel.compactionType
- val startTime = System.nanoTime();
+ val startTime = System.nanoTime()
val mergedLoadName = CarbonDataMergerUtil.getMergedLoadName(loadsToMerge)
var finalMergeStatus = false
val schemaName: String = carbonLoadModel.getDatabaseName
val factTableName = carbonLoadModel.getTableName
val validSegments: Array[String] = CarbonDataMergerUtil
.getValidSegments(loadsToMerge).split(',')
- val mergeLoadStartTime = CarbonLoaderUtil.readCurrentTime();
+ val mergeLoadStartTime = CarbonLoaderUtil.readCurrentTime()
val carbonMergerMapping = CarbonMergerMapping(storeLocation,
storePath,
partitioner,
- carbonTable.getMetaDataFilepath(),
+ carbonTable.getMetaDataFilepath,
mergedLoadName,
kettleHomePath,
cubeCreationTime,
@@ -82,19 +82,19 @@ object Compactor {
)
)
carbonLoadModel.setLoadMetadataDetails(segmentStatusManager
- .readLoadMetadata(carbonTable.getMetaDataFilepath()).toList.asJava
+ .readLoadMetadata(carbonTable.getMetaDataFilepath).toList.asJava
)
var execInstance = "1"
// in case of non dynamic executor allocation, number of executors are fixed.
if (sc.sparkContext.getConf.contains("spark.executor.instances")) {
execInstance = sc.sparkContext.getConf.get("spark.executor.instances")
- logger.info("spark.executor.instances property is set to =" + execInstance)
+ logger.info(s"spark.executor.instances property is set to = $execInstance")
} // in case of dynamic executor allocation, taking the max executors of the dynamic allocation.
else if (sc.sparkContext.getConf.contains("spark.dynamicAllocation.enabled")) {
if (sc.sparkContext.getConf.get("spark.dynamicAllocation.enabled").trim
.equalsIgnoreCase("true")) {
execInstance = sc.sparkContext.getConf.get("spark.dynamicAllocation.maxExecutors")
- logger.info("spark.dynamicAllocation.maxExecutors property is set to =" + execInstance)
+ logger.info(s"spark.dynamicAllocation.maxExecutors property is set to = $execInstance")
}
}
@@ -106,51 +106,38 @@ object Compactor {
execInstance
).collect
- if(mergeStatus.length == 0) {
+ if (mergeStatus.length == 0) {
finalMergeStatus = false
- }
- else {
+ } else {
finalMergeStatus = mergeStatus.forall(_._2)
}
if (finalMergeStatus) {
- val endTime = System.nanoTime();
- logger.info("time taken to merge " + mergedLoadName + " is " + (endTime - startTime))
+ val endTime = System.nanoTime()
+ logger.info(s"time taken to merge $mergedLoadName is ${ endTime - startTime }")
if (!CarbonDataMergerUtil
- .updateLoadMetadataWithMergeStatus(loadsToMerge, carbonTable.getMetaDataFilepath(),
+ .updateLoadMetadataWithMergeStatus(loadsToMerge, carbonTable.getMetaDataFilepath,
mergedLoadName, carbonLoadModel, mergeLoadStartTime, compactionType
)) {
- logger
- .audit("Compaction request failed for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName
- )
- logger
- .error("Compaction request failed for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName
- )
- throw new Exception("Compaction failed to update metadata for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName)
- }
- else {
- logger
- .audit("Compaction request completed for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName
- )
- logger
- .info("Compaction request completed for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName
- )
+ logger.audit(s"Compaction request failed for table ${ carbonLoadModel.getDatabaseName }." +
+ s"${ carbonLoadModel.getTableName }")
+ logger.error(s"Compaction request failed for table ${ carbonLoadModel.getDatabaseName }." +
+ s"${ carbonLoadModel.getTableName }")
+ throw new Exception(s"Compaction failed to update metadata for table" +
+ s" ${ carbonLoadModel.getDatabaseName }." +
+ s"${ carbonLoadModel.getTableName }")
+ } else {
+ logger.audit(s"Compaction request completed for table " +
+ s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ logger.info("Compaction request completed for table ${ carbonLoadModel.getDatabaseName } " +
+ s".${ carbonLoadModel.getTableName }")
}
- }
- else {
- logger
- .audit("Compaction request failed for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName
- )
- logger
- .error("Compaction request failed for table " + carbonLoadModel
- .getDatabaseName + "." + carbonLoadModel.getTableName
- )
+ } else {
+ logger.audit("Compaction request failed for table ${ carbonLoadModel.getDatabaseName } " +
+ s".${ carbonLoadModel.getTableName }"
+ )
+ logger.error("Compaction request failed for table ${ carbonLoadModel.getDatabaseName } " +
+ s".${ carbonLoadModel.getTableName }")
throw new Exception("Compaction Failure in Merger Rdd.")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
index d1b8bf3..e23b58d 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/tasks/DictionaryWriterTask.scala
@@ -23,6 +23,7 @@ import scala.collection.mutable
import org.apache.carbondata.common.factory.CarbonCommonFactory
import org.apache.carbondata.core.cache.dictionary.Dictionary
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.DataTypeUtil
import org.apache.carbondata.core.writer.CarbonDictionaryWriter
import org.apache.carbondata.spark.rdd.DictionaryLoadModel
@@ -63,8 +64,7 @@ class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
if (values.length >= 1) {
if (model.dictFileExists(columnIndex)) {
for (value <- values) {
- val parsedValue = org.apache.carbondata.core.util.DataTypeUtil
- .normalizeColumnValueForItsDataType(value,
+ val parsedValue = DataTypeUtil.normalizeColumnValueForItsDataType(value,
model.primDimensions(columnIndex))
if (null != parsedValue && dictionary.getSurrogateKey(parsedValue) ==
CarbonCommonConstants.INVALID_SURROGATE_KEY) {
@@ -75,8 +75,7 @@ class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
} else {
for (value <- values) {
- val parsedValue = org.apache.carbondata.core.util.DataTypeUtil
- .normalizeColumnValueForItsDataType(value,
+ val parsedValue = DataTypeUtil.normalizeColumnValueForItsDataType(value,
model.primDimensions(columnIndex))
if (null != parsedValue) {
writer.write(parsedValue)
@@ -88,8 +87,7 @@ class DictionaryWriterTask(valuesBuffer: mutable.HashSet[String],
} catch {
case ex: IOException =>
throw ex
- }
- finally {
+ } finally {
if (null != writer) {
writer.close()
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/thriftserver/CarbonThriftServer.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/thriftserver/CarbonThriftServer.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/thriftserver/CarbonThriftServer.scala
index 30d5871..efedc91 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/thriftserver/CarbonThriftServer.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/thriftserver/CarbonThriftServer.scala
@@ -45,7 +45,7 @@ object CarbonThriftServer {
} catch {
case e: Exception =>
val LOG = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
- LOG.error("Wrong value for carbon.spark.warmUpTime " + warmUpTime +
+ LOG.error(s"Wrong value for carbon.spark.warmUpTime $warmUpTime " +
"Using default Value and proceeding")
Thread.sleep(30000)
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index 8b89f5d..d5051cf 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -36,7 +36,7 @@ object CommonUtil {
if (noDictionaryDims.contains(x)) {
throw new MalformedCarbonCommandException(
"Column group is not supported for no dictionary columns:" + x)
- } else if (msrs.filter { msr => msr.column.equals(x) }.nonEmpty) {
+ } else if (msrs.exists(msr => msr.column.equals(x))) {
// if column is measure
throw new MalformedCarbonCommandException("Column group is not supported for measures:" + x)
} else if (foundIndExistingColGrp(x)) {
@@ -47,11 +47,11 @@ object CommonUtil {
} else if (isTimeStampColumn(x, dims)) {
throw new MalformedCarbonCommandException(
"Column group doesn't support Timestamp datatype:" + x)
- }
- // if invalid column is present
- else if (dims.filter { dim => dim.column.equalsIgnoreCase(x) }.isEmpty) {
+ }// if invalid column is
+ else if (!dims.exists(dim => dim.column.equalsIgnoreCase(x))) {
+ // present
throw new MalformedCarbonCommandException(
- "column in column group is not a valid column :" + x
+ "column in column group is not a valid column: " + x
)
}
}
@@ -110,7 +110,7 @@ object CommonUtil {
key.length()), value))
}
}
- if (fieldProps.isEmpty()) {
+ if (fieldProps.isEmpty) {
None
} else {
Some(fieldProps)
@@ -211,7 +211,7 @@ object CommonUtil {
var tableBlockSize: Integer = 0
if (tableProperties.get(CarbonCommonConstants.TABLE_BLOCKSIZE).isDefined) {
val blockSizeStr: String =
- parsePropertyValueStringInMB(tableProperties.get(CarbonCommonConstants.TABLE_BLOCKSIZE).get)
+ parsePropertyValueStringInMB(tableProperties(CarbonCommonConstants.TABLE_BLOCKSIZE))
try {
tableBlockSize = Integer.parseInt(blockSizeStr)
} catch {
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/util/DataTypeConverterUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/DataTypeConverterUtil.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/DataTypeConverterUtil.scala
index 1fee428..aa8fcd5 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/DataTypeConverterUtil.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/DataTypeConverterUtil.scala
@@ -35,7 +35,7 @@ object DataTypeConverterUtil {
case "timestamp" => DataType.TIMESTAMP
case "array" => DataType.ARRAY
case "struct" => DataType.STRUCT
- case _ => sys.error("Unsupported data type : " + dataType)
+ case _ => sys.error(s"Unsupported data type: $dataType")
}
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
index bd295bc..db01367 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/util/GlobalDictionaryUtil.scala
@@ -69,9 +69,9 @@ object GlobalDictionaryUtil extends Logging {
/**
* find columns which need to generate global dictionary.
*
- * @param dimensions dimension list of schema
- * @param headers column headers
- * @param columns column list of csv file
+ * @param dimensions dimension list of schema
+ * @param headers column headers
+ * @param columns column list of csv file
*/
def pruneDimensions(dimensions: Array[CarbonDimension],
headers: Array[String],
@@ -96,10 +96,10 @@ object GlobalDictionaryUtil extends Logging {
/**
* use this method to judge whether CarbonDimension use some encoding or not
- *
- * @param dimension carbonDimension
- * @param encoding the coding way of dimension
- * @param excludeEncoding the coding way to exclude
+ *
+ * @param dimension carbonDimension
+ * @param encoding the coding way of dimension
+ * @param excludeEncoding the coding way to exclude
*/
def hasEncoding(dimension: CarbonDimension,
encoding: Encoding,
@@ -129,7 +129,7 @@ object GlobalDictionaryUtil extends Logging {
if (dimension.hasEncoding(encoding) &&
(excludeEncoding == null || !dimension.hasEncoding(excludeEncoding))) {
if ((forPreDefDict && carbonLoadModel.getPredefDictFilePath(dimension) != null) ||
- (!forPreDefDict && carbonLoadModel.getPredefDictFilePath(dimension) == null)) {
+ (!forPreDefDict && carbonLoadModel.getPredefDictFilePath(dimension) == null)) {
dimensionsWithEncoding += dimension
}
}
@@ -137,8 +137,8 @@ object GlobalDictionaryUtil extends Logging {
}
def getPrimDimensionWithDict(carbonLoadModel: CarbonLoadModel,
- dimension: CarbonDimension,
- forPreDefDict: Boolean): Array[CarbonDimension] = {
+ dimension: CarbonDimension,
+ forPreDefDict: Boolean): Array[CarbonDimension] = {
val dimensionsWithDict = new ArrayBuffer[CarbonDimension]
gatherDimensionByEncoding(carbonLoadModel, dimension, Encoding.DICTIONARY,
Encoding.DIRECT_DICTIONARY,
@@ -267,19 +267,20 @@ object GlobalDictionaryUtil extends Logging {
}
def isHighCardinalityColumn(columnCardinality: Int,
- rowCount: Long,
- model: DictionaryLoadModel): Boolean = {
- (columnCardinality > model.highCardThreshold) && (rowCount > 0) &&
- (columnCardinality.toDouble / rowCount * 100 > model.rowCountPercentage)
+ rowCount: Long,
+ model: DictionaryLoadModel): Boolean = {
+ (columnCardinality > model.highCardThreshold) &&
+ (rowCount > 0) &&
+ (columnCardinality.toDouble / rowCount * 100 > model.rowCountPercentage)
}
/**
* create a instance of DictionaryLoadModel
*
- * @param carbonLoadModel carbon load model
- * @param table CarbonTableIdentifier
- * @param dimensions column list
- * @param hdfsLocation store location in HDFS
+ * @param carbonLoadModel carbon load model
+ * @param table CarbonTableIdentifier
+ * @param dimensions column list
+ * @param hdfsLocation store location in HDFS
* @param dictfolderPath path of dictionary folder
*/
def createDictionaryLoadModel(carbonLoadModel: CarbonLoadModel,
@@ -300,7 +301,7 @@ object GlobalDictionaryUtil extends Logging {
val carbonTablePath = CarbonStorePath.getCarbonTablePath(hdfsLocation, table)
val primDimensions = primDimensionsBuffer.map { x => x }.toArray
val dictDetail = CarbonSparkFactory.getDictionaryDetailService().
- getDictionaryDetail(dictfolderPath, primDimensions, table, hdfsLocation)
+ getDictionaryDetail(dictfolderPath, primDimensions, table, hdfsLocation)
val dictFilePaths = dictDetail.dictFilePaths
val dictFileExists = dictDetail.dictFileExists
val columnIdentifier = dictDetail.columnIdentifiers
@@ -311,14 +312,14 @@ object GlobalDictionaryUtil extends Logging {
val zookeeperUrl = CarbonProperties.getInstance.getProperty(CarbonCommonConstants.ZOOKEEPER_URL)
// load high cardinality identify configure
val highCardIdentifyEnable = CarbonProperties.getInstance().getProperty(
- CarbonCommonConstants.HIGH_CARDINALITY_IDENTIFY_ENABLE,
- CarbonCommonConstants.HIGH_CARDINALITY_IDENTIFY_ENABLE_DEFAULT).toBoolean
+ CarbonCommonConstants.HIGH_CARDINALITY_IDENTIFY_ENABLE,
+ CarbonCommonConstants.HIGH_CARDINALITY_IDENTIFY_ENABLE_DEFAULT).toBoolean
val highCardThreshold = CarbonProperties.getInstance().getProperty(
- CarbonCommonConstants.HIGH_CARDINALITY_THRESHOLD,
- CarbonCommonConstants.HIGH_CARDINALITY_THRESHOLD_DEFAULT).toInt
+ CarbonCommonConstants.HIGH_CARDINALITY_THRESHOLD,
+ CarbonCommonConstants.HIGH_CARDINALITY_THRESHOLD_DEFAULT).toInt
val rowCountPercentage = CarbonProperties.getInstance().getProperty(
- CarbonCommonConstants.HIGH_CARDINALITY_IN_ROW_COUNT_PERCENTAGE,
- CarbonCommonConstants.HIGH_CARDINALITY_IN_ROW_COUNT_PERCENTAGE_DEFAULT).toDouble
+ CarbonCommonConstants.HIGH_CARDINALITY_IN_ROW_COUNT_PERCENTAGE,
+ CarbonCommonConstants.HIGH_CARDINALITY_IN_ROW_COUNT_PERCENTAGE_DEFAULT).toDouble
// get load count
if (null == carbonLoadModel.getLoadMetadataDetails) {
@@ -346,8 +347,8 @@ object GlobalDictionaryUtil extends Logging {
/**
* load CSV files to DataFrame by using datasource "com.databricks.spark.csv"
*
- * @param sqlContext SQLContext
- * @param carbonLoadModel carbon data load model
+ * @param sqlContext SQLContext
+ * @param carbonLoadModel carbon data load model
*/
def loadDataFrame(sqlContext: SQLContext,
carbonLoadModel: CarbonLoadModel): DataFrame = {
@@ -356,16 +357,14 @@ object GlobalDictionaryUtil extends Logging {
.option("header", {
if (StringUtils.isEmpty(carbonLoadModel.getCsvHeader)) {
"true"
- }
- else {
+ } else {
"false"
}
})
.option("delimiter", {
if (StringUtils.isEmpty(carbonLoadModel.getCsvDelimiter)) {
"" + DEFAULT_SEPARATOR
- }
- else {
+ } else {
carbonLoadModel.getCsvDelimiter
}
})
@@ -377,8 +376,7 @@ object GlobalDictionaryUtil extends Logging {
.option("quote", {
if (StringUtils.isEmpty(carbonLoadModel.getQuoteChar)) {
"" + DEFAULT_QUOTE_CHARACTER
- }
- else {
+ } else {
carbonLoadModel.getQuoteChar
}
})
@@ -388,9 +386,9 @@ object GlobalDictionaryUtil extends Logging {
}
private def updateTableMetadata(carbonLoadModel: CarbonLoadModel,
- sqlContext: SQLContext,
- model: DictionaryLoadModel,
- noDictDimension: Array[CarbonDimension]): Unit = {
+ sqlContext: SQLContext,
+ model: DictionaryLoadModel,
+ noDictDimension: Array[CarbonDimension]): Unit = {
val carbonTablePath = CarbonStorePath.getCarbonTablePath(model.hdfsLocation,
model.table)
@@ -413,12 +411,11 @@ object GlobalDictionaryUtil extends Logging {
// update Metadata
val catalog = CarbonEnv.getInstance(sqlContext).carbonCatalog
catalog.updateMetadataByThriftTable(schemaFilePath, tableInfo,
- model.table.getDatabaseName, model.table.getTableName, carbonLoadModel.getStorePath)
+ model.table.getDatabaseName, model.table.getTableName, carbonLoadModel.getStorePath)
// update CarbonDataLoadSchema
val carbonTable = catalog.lookupRelation1(Option(model.table.getDatabaseName),
- model.table.getTableName)(sqlContext)
- .asInstanceOf[CarbonRelation].tableMeta.carbonTable
+ model.table.getTableName)(sqlContext).asInstanceOf[CarbonRelation].tableMeta.carbonTable
carbonLoadModel.setCarbonDataLoadSchema(new CarbonDataLoadSchema(carbonTable))
}
@@ -429,9 +426,9 @@ object GlobalDictionaryUtil extends Logging {
* @param status checking whether the generating is successful
*/
private def checkStatus(carbonLoadModel: CarbonLoadModel,
- sqlContext: SQLContext,
- model: DictionaryLoadModel,
- status: Array[(Int, String, Boolean)]) = {
+ sqlContext: SQLContext,
+ model: DictionaryLoadModel,
+ status: Array[(Int, String, Boolean)]) = {
var result = false
val noDictionaryColumns = new ArrayBuffer[CarbonDimension]
val tableName = model.table.getTableName
@@ -441,7 +438,9 @@ object GlobalDictionaryUtil extends Logging {
result = true
logError(s"table:$tableName column:$columnName generate global dictionary file failed")
}
- if (x._3) noDictionaryColumns += model.primDimensions(x._1)
+ if (x._3) {
+ noDictionaryColumns += model.primDimensions(x._1)
+ }
}
if (noDictionaryColumns.nonEmpty) {
updateTableMetadata(carbonLoadModel, sqlContext, model, noDictionaryColumns.toArray)
@@ -457,23 +456,23 @@ object GlobalDictionaryUtil extends Logging {
/**
* get external columns and whose dictionary file path
*
- * @param colDictFilePath external column dict file path
- * @param table table identifier
- * @param dimensions dimension columns
+ * @param colDictFilePath external column dict file path
+ * @param table table identifier
+ * @param dimensions dimension columns
*/
private def setPredefinedColumnDictPath(carbonLoadModel: CarbonLoadModel,
- colDictFilePath: String,
- table: CarbonTableIdentifier,
- dimensions: Array[CarbonDimension]) = {
+ colDictFilePath: String,
+ table: CarbonTableIdentifier,
+ dimensions: Array[CarbonDimension]) = {
val colFileMapArray = colDictFilePath.split(",")
for (colPathMap <- colFileMapArray) {
val colPathMapTrim = colPathMap.trim
val colNameWithPath = colPathMapTrim.split(":")
if (colNameWithPath.length == 1) {
logError("the format of external column dictionary should be " +
- "columnName:columnPath, please check")
+ "columnName:columnPath, please check")
throw new DataLoadingException("the format of predefined column dictionary" +
- " should be columnName:columnPath, please check")
+ " should be columnName:columnPath, please check")
}
setPredefineDict(carbonLoadModel, dimensions, table, colNameWithPath(0),
FileUtils.getPaths(colPathMapTrim.substring(colNameWithPath(0).length + 1)))
@@ -482,26 +481,25 @@ object GlobalDictionaryUtil extends Logging {
/**
* set pre defined dictionary for dimension
- *
- * @param dimensions all the dimensions
- * @param table carbon table identifier
- * @param colName user specified column name for predefined dict
- * @param colDictPath column dictionary file path
- * @param parentDimName parent dimenion for complex type
+ *
+ * @param dimensions all the dimensions
+ * @param table carbon table identifier
+ * @param colName user specified column name for predefined dict
+ * @param colDictPath column dictionary file path
+ * @param parentDimName parent dimenion for complex type
*/
def setPredefineDict(carbonLoadModel: CarbonLoadModel,
- dimensions: Array[CarbonDimension],
- table: CarbonTableIdentifier,
- colName: String,
- colDictPath: String,
- parentDimName: String = "") {
+ dimensions: Array[CarbonDimension],
+ table: CarbonTableIdentifier,
+ colName: String,
+ colDictPath: String,
+ parentDimName: String = "") {
val middleDimName = colName.split("\\.")(0)
val dimParent = parentDimName + {
colName match {
case "" => colName
case _ =>
- if (parentDimName.isEmpty) middleDimName
- else "." + middleDimName
+ if (parentDimName.isEmpty) middleDimName else "." + middleDimName
}
}
// judge whether the column is exists
@@ -509,9 +507,10 @@ object GlobalDictionaryUtil extends Logging {
_.getColName.equalsIgnoreCase(dimParent))
if (preDictDimensionOption.length == 0) {
logError(s"Column $dimParent is not a key column " +
- s"in ${table.getDatabaseName}.${table.getTableName}")
+ s"in ${ table.getDatabaseName }.${ table.getTableName }")
throw new DataLoadingException(s"Column $dimParent is not a key column. " +
- s"Only key column can be part of dictionary and used in COLUMNDICT option.")
+ s"Only key column can be part of dictionary " +
+ s"and used in COLUMNDICT option.")
}
val preDictDimension = preDictDimensionOption(0)
if (preDictDimension.isComplex) {
@@ -520,9 +519,11 @@ object GlobalDictionaryUtil extends Logging {
val currentColName = {
preDictDimension.getDataType match {
case DataType.ARRAY =>
- if (children(0).isComplex) "val." +
- colName.substring(middleDimName.length + 1)
- else "val"
+ if (children(0).isComplex) {
+ "val." + colName.substring(middleDimName.length + 1)
+ } else {
+ "val"
+ }
case _ => colName.substring(middleDimName.length + 1)
}
}
@@ -534,27 +535,27 @@ object GlobalDictionaryUtil extends Logging {
}
/**
- * use external dimension column to generate global dictionary
+ * use external dimension column to generate global dictionary
*
- * @param colDictFilePath external column dict file path
- * @param table table identifier
- * @param dimensions dimension column
- * @param carbonLoadModel carbon load model
- * @param sqlContext spark sql context
- * @param hdfsLocation store location on hdfs
+ * @param colDictFilePath external column dict file path
+ * @param table table identifier
+ * @param dimensions dimension column
+ * @param carbonLoadModel carbon load model
+ * @param sqlContext spark sql context
+ * @param hdfsLocation store location on hdfs
* @param dictFolderPath generated global dict file path
*/
private def generatePredefinedColDictionary(colDictFilePath: String,
- table: CarbonTableIdentifier,
- dimensions: Array[CarbonDimension],
- carbonLoadModel: CarbonLoadModel,
- sqlContext: SQLContext,
- hdfsLocation: String,
- dictFolderPath: String) = {
+ table: CarbonTableIdentifier,
+ dimensions: Array[CarbonDimension],
+ carbonLoadModel: CarbonLoadModel,
+ sqlContext: SQLContext,
+ hdfsLocation: String,
+ dictFolderPath: String) = {
// set pre defined dictionary column
setPredefinedColumnDictPath(carbonLoadModel, colDictFilePath, table, dimensions)
val dictLoadModel = createDictionaryLoadModel(carbonLoadModel, table, dimensions,
- hdfsLocation, dictFolderPath, true)
+ hdfsLocation, dictFolderPath, forPreDefDict = true)
// new RDD to achieve distributed column dict generation
val extInputRDD = new CarbonColumnDictGenerateRDD(carbonLoadModel, dictLoadModel,
sqlContext.sparkContext, table, dimensions, hdfsLocation, dictFolderPath)
@@ -600,7 +601,7 @@ object GlobalDictionaryUtil extends Logging {
* @param csvFileColumns
*/
private def parseRecord(x: String, accum: Accumulator[Int],
- csvFileColumns: Array[String]) : (String, String) = {
+ csvFileColumns: Array[String]): (String, String) = {
val tokens = x.split("" + DEFAULT_SEPARATOR)
var columnName: String = ""
var value: String = ""
@@ -610,7 +611,7 @@ object GlobalDictionaryUtil extends Logging {
accum += 1
} else if (tokens.size == 1) {
// such as "1", "jone", throw ex
- if (x.contains(",") == false) {
+ if (!x.contains(",")) {
accum += 1
} else {
try {
@@ -644,10 +645,10 @@ object GlobalDictionaryUtil extends Logging {
* @return allDictionaryRdd
*/
private def readAllDictionaryFiles(sqlContext: SQLContext,
- csvFileColumns: Array[String],
- requireColumns: Array[String],
- allDictionaryPath: String,
- accumulator: Accumulator[Int]) = {
+ csvFileColumns: Array[String],
+ requireColumns: Array[String],
+ allDictionaryPath: String,
+ accumulator: Accumulator[Int]) = {
var allDictionaryRdd: RDD[(String, Iterable[String])] = null
try {
// read local dictionary file, and spilt (columnIndex, columnValue)
@@ -686,7 +687,7 @@ object GlobalDictionaryUtil extends Logging {
true
} else {
logWarning("No dictionary files found or empty dictionary files! " +
- "Won't generate new dictionary.")
+ "Won't generate new dictionary.")
false
}
} else {
@@ -699,7 +700,7 @@ object GlobalDictionaryUtil extends Logging {
true
} else {
logWarning("No dictionary files found or empty dictionary files! " +
- "Won't generate new dictionary.")
+ "Won't generate new dictionary.")
false
}
} else {
@@ -726,7 +727,7 @@ object GlobalDictionaryUtil extends Logging {
} else {
carbonLoadModel.getCsvDelimiter
}
- headers = readLine.toLowerCase().split(delimiter);
+ headers = readLine.toLowerCase().split(delimiter)
} else {
logError("Not found file header! Please set fileheader")
throw new IOException("Failed to get file header")
@@ -737,13 +738,13 @@ object GlobalDictionaryUtil extends Logging {
/**
* generate global dictionary with SQLContext and CarbonLoadModel
*
- * @param sqlContext sql context
- * @param carbonLoadModel carbon load model
+ * @param sqlContext sql context
+ * @param carbonLoadModel carbon load model
*/
def generateGlobalDictionary(sqlContext: SQLContext,
- carbonLoadModel: CarbonLoadModel,
- storePath: String,
- dataFrame: Option[DataFrame] = None): Unit = {
+ carbonLoadModel: CarbonLoadModel,
+ storePath: String,
+ dataFrame: Option[DataFrame] = None): Unit = {
try {
var carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
var carbonTableIdentifier = carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
@@ -757,7 +758,7 @@ object GlobalDictionaryUtil extends Logging {
carbonLoadModel.initPredefDictMap()
val allDictionaryPath = carbonLoadModel.getAllDictPath
- if(StringUtils.isEmpty(allDictionaryPath)) {
+ if (StringUtils.isEmpty(allDictionaryPath)) {
logInfo("Generate global dictionary from source data files!")
// load data by using dataSource com.databricks.spark.csv
var df = if (dataFrame.isDefined) {
@@ -767,8 +768,7 @@ object GlobalDictionaryUtil extends Logging {
}
var headers = if (StringUtils.isEmpty(carbonLoadModel.getCsvHeader)) {
df.columns
- }
- else {
+ } else {
carbonLoadModel.getCsvHeader.split("" + DEFAULT_SEPARATOR)
}
headers = headers.map(headerName => headerName.trim)
@@ -779,8 +779,9 @@ object GlobalDictionaryUtil extends Logging {
dimensions, carbonLoadModel, sqlContext, storePath, dictfolderPath)
}
if (headers.length > df.columns.length) {
- val msg = "The number of columns in the file header do not match the number of " +
- "columns in the data file; Either delimiter or fileheader provided is not correct"
+ val msg = "The number of columns in the file header do not match the " +
+ "number of columns in the data file; Either delimiter " +
+ "or fileheader provided is not correct"
logError(msg)
throw new DataLoadingException(msg)
}
@@ -829,7 +830,7 @@ object GlobalDictionaryUtil extends Logging {
} else {
logInfo("Generate global dictionary from dictionary files!")
val isNonempty = validateAllDictionaryPath(allDictionaryPath)
- if(isNonempty) {
+ if (isNonempty) {
var headers = if (StringUtils.isEmpty(carbonLoadModel.getCsvHeader)) {
getHeaderFormFactFile(carbonLoadModel)
} else {
@@ -837,8 +838,7 @@ object GlobalDictionaryUtil extends Logging {
}
headers = headers.map(headerName => headerName.trim)
// prune columns according to the CSV file header, dimension columns
- val (requireDimension, requireColumnNames) =
- pruneDimensions(dimensions, headers, headers)
+ val (requireDimension, requireColumnNames) = pruneDimensions(dimensions, headers, headers)
if (requireDimension.nonEmpty) {
val model = createDictionaryLoadModel(carbonLoadModel, carbonTableIdentifier,
requireDimension, storePath, dictfolderPath, false)
@@ -857,7 +857,7 @@ object GlobalDictionaryUtil extends Logging {
// if the dictionary contains wrong format record, throw ex
if (accumulator.value > 0) {
throw new DataLoadingException("Data Loading failure, dictionary values are " +
- "not in correct format!")
+ "not in correct format!")
}
} else {
logInfo("have no column need to generate global dictionary")
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
index e27d166..a792eca 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonContext.scala
@@ -44,14 +44,14 @@ class CarbonContext(
metaStorePath: String) extends HiveContext(sc) with Logging {
self =>
- def this (sc: SparkContext) = {
- this (sc,
+ def this(sc: SparkContext) = {
+ this(sc,
new File(CarbonCommonConstants.STORE_LOCATION_DEFAULT_VAL).getCanonicalPath,
new File(CarbonCommonConstants.METASTORE_LOCATION_DEFAULT_VAL).getCanonicalPath)
}
- def this (sc: SparkContext, storePath: String) = {
- this (sc,
+ def this(sc: SparkContext, storePath: String) = {
+ this(sc,
storePath,
new File(CarbonCommonConstants.METASTORE_LOCATION_DEFAULT_VAL).getCanonicalPath)
}
@@ -107,9 +107,9 @@ class CarbonContext(
if (sc.hadoopConfiguration.get(CarbonCommonConstants.HIVE_CONNECTION_URL) == null) {
val metaStorePathAbsolute = new File(metaStorePath).getCanonicalPath
val hiveMetaStoreDB = metaStorePathAbsolute + "/metastore_db"
- logDebug(s"metastore db is going to be created in location : $hiveMetaStoreDB")
+ logDebug(s"metastore db is going to be created in location: $hiveMetaStoreDB")
super.configure() ++ Map((CarbonCommonConstants.HIVE_CONNECTION_URL,
- s"jdbc:derby:;databaseName=$hiveMetaStoreDB;create=true"),
+ s"jdbc:derby:;databaseName=$hiveMetaStoreDB;create=true"),
("hive.metastore.warehouse.dir", metaStorePathAbsolute + "/hivemetadata"))
} else {
super.configure()
@@ -186,7 +186,7 @@ object CarbonContext {
/**
*
* Requesting the extra executors other than the existing ones.
- *
+ *
* @param sc
* @param numExecutors
* @return
@@ -194,12 +194,12 @@ object CarbonContext {
final def ensureExecutors(sc: SparkContext, numExecutors: Int): Boolean = {
sc.schedulerBackend match {
case b: CoarseGrainedSchedulerBackend =>
- val requiredExecutors = numExecutors - b.numExistingExecutors
+ val requiredExecutors = numExecutors - b.numExistingExecutors
LOGGER
- .info("number of executors is =" + numExecutors + " existing executors are =" + b
- .numExistingExecutors
+ .info(s"number of executors is =$numExecutors existing executors are =" +
+ s"${ b.numExistingExecutors }"
)
- if(requiredExecutors > 0) {
+ if (requiredExecutors > 0) {
b.requestExecutors(requiredExecutors)
}
true
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
index c1a8818..f4fe900 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala
@@ -111,10 +111,10 @@ case class CarbonDictionaryDecoder(
case DataType.TIMESTAMP => TimestampType
case DataType.STRUCT =>
CarbonMetastoreTypes
- .toDataType(s"struct<${ relation.getStructChildren(carbonDimension.getColName) }>")
+ .toDataType(s"struct<${ relation.getStructChildren(carbonDimension.getColName) }>")
case DataType.ARRAY =>
CarbonMetastoreTypes
- .toDataType(s"array<${ relation.getArrayChildren(carbonDimension.getColName) }>")
+ .toDataType(s"array<${ relation.getArrayChildren(carbonDimension.getColName) }>")
}
}
@@ -131,7 +131,7 @@ case class CarbonDictionaryDecoder(
carbonDimension.hasEncoding(Encoding.DICTIONARY) &&
!carbonDimension.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
(carbonTable.getFactTableName, carbonDimension.getColumnIdentifier,
- carbonDimension.getDataType)
+ carbonDimension.getDataType)
} else {
(null, null, null)
}
@@ -160,7 +160,7 @@ case class CarbonDictionaryDecoder(
(carbonTable.getFactTableName, carbonTable.getAbsoluteTableIdentifier)
}.toMap
- val recorder = CarbonTimeStatisticsFactory.createExecutorRecorder(queryId);
+ val recorder = CarbonTimeStatisticsFactory.createExecutorRecorder(queryId)
if (isRequiredToDecode) {
val dataTypes = child.output.map { attr => attr.dataType }
child.execute().mapPartitions { iter =>
@@ -186,7 +186,7 @@ case class CarbonDictionaryDecoder(
var total = 0L
override final def hasNext: Boolean = {
flag = iter.hasNext
- if (false == flag && total > 0) {
+ if (!flag && total > 0) {
val queryStatistic = new QueryStatistic()
queryStatistic
.addFixedTimeStatistic(QueryStatisticsConstants.PREPARE_RESULT, total)
@@ -202,8 +202,8 @@ case class CarbonDictionaryDecoder(
dictIndex.foreach { index =>
if (data(index) != null) {
data(index) = DataTypeUtil.getDataBasedOnDataType(dicts(index)
- .getDictionaryValueForKey(data(index).asInstanceOf[Int]),
- getDictionaryColumnIds(index)._3)
+ .getDictionaryValueForKey(data(index).asInstanceOf[Int]),
+ getDictionaryColumnIds(index)._3)
}
}
val result = unsafeProjection(new GenericMutableRow(data))
@@ -231,7 +231,7 @@ case class CarbonDictionaryDecoder(
if (f._2 != null) {
try {
cache.get(new DictionaryColumnUniqueIdentifier(
- atiMap.get(f._1).get.getCarbonTableIdentifier,
+ atiMap(f._1).getCarbonTableIdentifier,
f._2, f._3))
} catch {
case _: Throwable => null
@@ -242,4 +242,5 @@ case class CarbonDictionaryDecoder(
}
dicts
}
+
}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/6391c2be/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
index 6f149f7..f9a0a9d 100644
--- a/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
+++ b/integration/spark/src/main/scala/org/apache/spark/sql/CarbonSqlParser.scala
@@ -171,39 +171,48 @@ class CarbonSqlParser()
}
import lexical.Identifier
- implicit def regexToParser(regex: Regex): Parser[String] = acceptMatch(
- s"identifier matching regex ${regex}",
+
+ implicit def regexToParser(regex: Regex): Parser[String] = {
+ acceptMatch(
+ s"identifier matching regex ${ regex }",
{ case Identifier(str) if regex.unapplySeq(str).isDefined => str }
- )
- override def parse(input: String): LogicalPlan = synchronized {
- // Initialize the Keywords.
- initLexical
- phrase(start)(new lexical.Scanner(input)) match {
- case Success(plan, _) => plan match {
- case x: LoadTable =>
- x.inputSqlString = input
- x
- case logicalPlan => logicalPlan
+ )
+ }
+
+ override def parse(input: String): LogicalPlan = {
+ synchronized {
+ // Initialize the Keywords.
+ initLexical
+ phrase(start)(new lexical.Scanner(input)) match {
+ case Success(plan, _) => plan match {
+ case x: LoadTable =>
+ x.inputSqlString = input
+ x
+ case logicalPlan => logicalPlan
+ }
+ case failureOrError => sys.error(failureOrError.toString)
}
- case failureOrError => sys.error(failureOrError.toString)
}
}
/**
* This will convert key word to regular expression.
+ *
* @param keys
* @return
*/
- private def carbonKeyWord(keys: String) =
+ private def carbonKeyWord(keys: String) = {
("(?i)" + keys).r
+ }
override protected lazy val start: Parser[LogicalPlan] = explainPlan | startCommand
protected lazy val startCommand: Parser[LogicalPlan] = createDatabase | dropDatabase |
- loadManagement | describeTable | showLoads | alterTable | createTable
+ loadManagement | describeTable |
+ showLoads | alterTable | createTable
protected lazy val loadManagement: Parser[LogicalPlan] = deleteLoadsByID | deleteLoadsByLoadDate |
- cleanFiles | loadDataNew
+ cleanFiles | loadDataNew
protected val escapedIdentifier = "`([^`]+)`".r
@@ -248,9 +257,9 @@ class CarbonSqlParser()
var dimensions: Seq[Field] = Seq()
dims.foreach { dimension =>
dimension.dataType.getOrElse("NIL") match {
- case "Array" => complexDimensions = complexDimensions:+dimension
- case "Struct" => complexDimensions = complexDimensions:+dimension
- case _ => dimensions = dimensions:+dimension
+ case "Array" => complexDimensions = complexDimensions :+ dimension
+ case "Struct" => complexDimensions = complexDimensions :+ dimension
+ case _ => dimensions = dimensions :+ dimension
}
}
dimensions ++ complexDimensions
@@ -276,22 +285,22 @@ class CarbonSqlParser()
* For handling the create table DDl systax compatible to Hive syntax
*/
protected lazy val createTable: Parser[LogicalPlan] =
- restInput ^^ {
+ restInput ^^ {
- case statement =>
- try {
- // DDl will be parsed and we get the AST tree from the HiveQl
- val node = HiveQlWrapper.getAst(statement)
- // processing the AST tree
- nodeToPlan(node)
- } catch {
- // MalformedCarbonCommandException need to be throw directly, parser will catch it
- case ce: MalformedCarbonCommandException =>
- throw ce
- case e: Exception =>
- sys.error("Parsing error") // no need to do anything.
- }
- }
+ case statement =>
+ try {
+ // DDl will be parsed and we get the AST tree from the HiveQl
+ val node = HiveQlWrapper.getAst(statement)
+ // processing the AST tree
+ nodeToPlan(node)
+ } catch {
+ // MalformedCarbonCommandException need to be throw directly, parser will catch it
+ case ce: MalformedCarbonCommandException =>
+ throw ce
+ case e: Exception =>
+ sys.error("Parsing error") // no need to do anything.
+ }
+ }
private def getScaleAndPrecision(dataType: String): (Int, Int) = {
val m: Matcher = Pattern.compile("^decimal\\(([^)]+)\\)").matcher(dataType)
@@ -313,17 +322,17 @@ class CarbonSqlParser()
case Token("TOK_CREATETABLE", children) =>
- var fields: Seq[Field] = Seq[Field]()
- var tableComment: String = ""
- var tableProperties = Map[String, String]()
- var partitionCols: Seq[PartitionerField] = Seq[PartitionerField]()
- var likeTableName: String = ""
- var storedBy: String = ""
- var ifNotExistPresent: Boolean = false
- var dbName: Option[String] = None
- var tableName: String = ""
+ var fields: Seq[Field] = Seq[Field]()
+ var tableComment: String = ""
+ var tableProperties = Map[String, String]()
+ var partitionCols: Seq[PartitionerField] = Seq[PartitionerField]()
+ var likeTableName: String = ""
+ var storedBy: String = ""
+ var ifNotExistPresent: Boolean = false
+ var dbName: Option[String] = None
+ var tableName: String = ""
- try {
+ try {
// Checking whether create table request is carbon table
children.collect {
@@ -332,7 +341,7 @@ class CarbonSqlParser()
case _ =>
}
if (!(storedBy.equals(CarbonContext.datasourceName) ||
- storedBy.equals(CarbonContext.datasourceShortName))) {
+ storedBy.equals(CarbonContext.datasourceShortName))) {
sys.error("Not a carbon format request")
}
@@ -362,7 +371,7 @@ class CarbonSqlParser()
match {
case Success(field, _) => field
case failureOrError => throw new MalformedCarbonCommandException(
- s"Unsupported data type : $col.getType")
+ s"Unsupported data type: $col.getType")
}
// the data type of the decimal type will be like decimal(10,0)
// so checking the start of the string and taking the precision and scale.
@@ -405,7 +414,7 @@ class CarbonSqlParser()
if (repeatedProperties.nonEmpty) {
val repeatedPropStr: String = repeatedProperties.mkString(",")
throw new MalformedCarbonCommandException("Table properties is repeated: " +
- repeatedPropStr)
+ repeatedPropStr)
}
tableProperties ++= propertySeq
@@ -430,12 +439,17 @@ class CarbonSqlParser()
// get logical plan.
CreateTable(tableModel)
- }
- catch {
+ } catch {
case ce: MalformedCarbonCommandException =>
- val message = if (tableName.isEmpty) "Create table command failed. "
- else if (dbName.isEmpty) s"Create table command failed for $tableName. "
- else s"Create table command failed for ${dbName.get}.$tableName. "
+ val message = if (tableName.isEmpty) {
+ "Create table command failed. "
+ }
+ else if (dbName.isEmpty) {
+ s"Create table command failed for $tableName. "
+ }
+ else {
+ s"Create table command failed for ${ dbName.get }.$tableName. "
+ }
LOGGER.audit(message + ce.getMessage)
throw ce
}
@@ -488,19 +502,24 @@ class CarbonSqlParser()
* @return
*/
protected def prepareTableModel(ifNotExistPresent: Boolean, dbName: Option[String]
- , tableName: String, fields: Seq[Field],
- partitionCols: Seq[PartitionerField],
- tableProperties: Map[String, String]): tableModel
+ , tableName: String, fields: Seq[Field],
+ partitionCols: Seq[PartitionerField],
+ tableProperties: Map[String, String]): tableModel
= {
val (dims: Seq[Field], noDictionaryDims: Seq[String]) = extractDimColsAndNoDictionaryFields(
fields, tableProperties)
if (dims.isEmpty) {
- throw new MalformedCarbonCommandException(s"Table ${dbName.getOrElse(
- CarbonCommonConstants.DATABASE_DEFAULT_NAME)}.$tableName"
- + " can not be created without key columns. Please use DICTIONARY_INCLUDE or " +
- "DICTIONARY_EXCLUDE to set at least one key column " +
- "if all specified columns are numeric types")
+ throw new MalformedCarbonCommandException(s"Table ${
+ dbName.getOrElse(
+ CarbonCommonConstants.DATABASE_DEFAULT_NAME)
+ }.$tableName"
+ +
+ " can not be created without key columns. Please " +
+ "use DICTIONARY_INCLUDE or " +
+ "DICTIONARY_EXCLUDE to set at least one key " +
+ "column " +
+ "if all specified columns are numeric types")
}
val msrs: Seq[Field] = extractMsrColsFromFields(fields, tableProperties)
@@ -508,7 +527,7 @@ class CarbonSqlParser()
val colProps = extractColumnProperties(fields, tableProperties)
// get column groups configuration from table properties.
val groupCols: Seq[String] = updateColumnGroupsInField(tableProperties,
- noDictionaryDims, msrs, dims)
+ noDictionaryDims, msrs, dims)
// get no inverted index columns from table properties.
val noInvertedIdxCols = extractNoInvertedIndexColumns(fields, tableProperties)
@@ -554,8 +573,7 @@ class CarbonSqlParser()
}
// This will be furthur handled.
CommonUtil.arrangeColGrpsInSchemaOrder(splittedColGrps, dims)
- }
- else {
+ } else {
null
}
}
@@ -603,7 +621,7 @@ class CarbonSqlParser()
* @return
*/
protected def getPartitionerObject(partitionCols: Seq[PartitionerField],
- tableProperties: Map[String, String]):
+ tableProperties: Map[String, String]):
Option[Partitioner] = {
// by default setting partition class empty.
@@ -649,8 +667,8 @@ class CarbonSqlParser()
}
protected def fillAllChildrenColumnProperty(parent: String, fieldChildren: Option[List[Field]],
- tableProperties: Map[String, String],
- colPropMap: java.util.HashMap[String, java.util.List[ColumnProperty]]) {
+ tableProperties: Map[String, String],
+ colPropMap: java.util.HashMap[String, java.util.List[ColumnProperty]]) {
fieldChildren.foreach(fields => {
fields.foreach(field => {
fillColumnProperty(Some(parent), field.column, tableProperties, colPropMap)
@@ -661,9 +679,9 @@ class CarbonSqlParser()
}
protected def fillColumnProperty(parentColumnName: Option[String],
- columnName: String,
- tableProperties: Map[String, String],
- colPropMap: java.util.HashMap[String, java.util.List[ColumnProperty]]) {
+ columnName: String,
+ tableProperties: Map[String, String],
+ colPropMap: java.util.HashMap[String, java.util.List[ColumnProperty]]) {
val (tblPropKey, colProKey) = getKey(parentColumnName, columnName)
val colProps = CommonUtil.getColumnProperties(tblPropKey, tableProperties)
if (colProps.isDefined) {
@@ -672,7 +690,7 @@ class CarbonSqlParser()
}
def getKey(parentColumnName: Option[String],
- columnName: String): (String, String) = {
+ columnName: String): (String, String) = {
if (parentColumnName.isDefined) {
if (columnName == "val") {
(parentColumnName.get, parentColumnName.get + "." + columnName)
@@ -683,6 +701,7 @@ class CarbonSqlParser()
(columnName, columnName)
}
}
+
/**
* This will extract the no inverted columns fields.
* By default all dimensions use inverted index.
@@ -692,7 +711,7 @@ class CarbonSqlParser()
* @return
*/
protected def extractNoInvertedIndexColumns(fields: Seq[Field],
- tableProperties: Map[String, String]):
+ tableProperties: Map[String, String]):
Seq[String] = {
// check whether the column name is in fields
var noInvertedIdxColsProps: Array[String] = Array[String]()
@@ -703,17 +722,17 @@ class CarbonSqlParser()
tableProperties.get("NO_INVERTED_INDEX").get.split(',').map(_.trim)
noInvertedIdxColsProps
.map { noInvertedIdxColProp =>
- if (!fields.exists(x => x.column.equalsIgnoreCase(noInvertedIdxColProp))) {
- val errormsg = "NO_INVERTED_INDEX column: " + noInvertedIdxColProp +
- " does not exist in table. Please check create table statement."
- throw new MalformedCarbonCommandException(errormsg)
+ if (!fields.exists(x => x.column.equalsIgnoreCase(noInvertedIdxColProp))) {
+ val errormsg = "NO_INVERTED_INDEX column: " + noInvertedIdxColProp +
+ " does not exist in table. Please check create table statement."
+ throw new MalformedCarbonCommandException(errormsg)
+ }
}
- }
}
// check duplicate columns and only 1 col left
val distinctCols = noInvertedIdxColsProps.toSet
// extract the no inverted index columns
- fields.foreach( field => {
+ fields.foreach(field => {
if (distinctCols.exists(x => x.equalsIgnoreCase(field.column))) {
noInvertedIdxCols :+= field.column
}
@@ -731,7 +750,7 @@ class CarbonSqlParser()
* @return
*/
protected def extractDimColsAndNoDictionaryFields(fields: Seq[Field],
- tableProperties: Map[String, String]):
+ tableProperties: Map[String, String]):
(Seq[Field], Seq[String]) = {
var dimFields: LinkedHashSet[Field] = LinkedHashSet[Field]()
var dictExcludeCols: Array[String] = Array[String]()
@@ -746,18 +765,18 @@ class CarbonSqlParser()
.map { dictExcludeCol =>
if (!fields.exists(x => x.column.equalsIgnoreCase(dictExcludeCol))) {
val errormsg = "DICTIONARY_EXCLUDE column: " + dictExcludeCol +
- " does not exist in table. Please check create table statement."
+ " does not exist in table. Please check create table statement."
throw new MalformedCarbonCommandException(errormsg)
} else {
- val dataType = fields.find (x =>
+ val dataType = fields.find(x =>
x.column.equalsIgnoreCase(dictExcludeCol)).get.dataType.get
if (isComplexDimDictionaryExclude(dataType)) {
val errormsg = "DICTIONARY_EXCLUDE is unsupported for complex datatype column: " +
- dictExcludeCol
+ dictExcludeCol
throw new MalformedCarbonCommandException(errormsg)
} else if (!isStringAndTimestampColDictionaryExclude(dataType)) {
val errorMsg = "DICTIONARY_EXCLUDE is unsupported for " + dataType.toLowerCase() +
- " data type column: " + dictExcludeCol
+ " data type column: " + dictExcludeCol
throw new MalformedCarbonCommandException(errorMsg)
}
}
@@ -768,19 +787,19 @@ class CarbonSqlParser()
dictIncludeCols =
tableProperties.get(CarbonCommonConstants.DICTIONARY_INCLUDE).get.split(",").map(_.trim)
dictIncludeCols.map { distIncludeCol =>
- if (!fields.exists(x => x.column.equalsIgnoreCase(distIncludeCol.trim))) {
- val errormsg = "DICTIONARY_INCLUDE column: " + distIncludeCol.trim +
- " does not exist in table. Please check create table statement."
- throw new MalformedCarbonCommandException(errormsg)
- }
+ if (!fields.exists(x => x.column.equalsIgnoreCase(distIncludeCol.trim))) {
+ val errormsg = "DICTIONARY_INCLUDE column: " + distIncludeCol.trim +
+ " does not exist in table. Please check create table statement."
+ throw new MalformedCarbonCommandException(errormsg)
}
+ }
}
// include cols should contain exclude cols
dictExcludeCols.foreach { dicExcludeCol =>
if (dictIncludeCols.exists(x => x.equalsIgnoreCase(dicExcludeCol))) {
val errormsg = "DICTIONARY_EXCLUDE can not contain the same column: " + dicExcludeCol +
- " with DICTIONARY_INCLUDE. Please check create table statement."
+ " with DICTIONARY_INCLUDE. Please check create table statement."
throw new MalformedCarbonCommandException(errormsg)
}
}
@@ -794,11 +813,9 @@ class CarbonSqlParser()
noDictionaryDims :+= field.column
}
dimFields += field
- }
- else if (dictIncludeCols.exists(x => x.equalsIgnoreCase(field.column))) {
+ } else if (dictIncludeCols.exists(x => x.equalsIgnoreCase(field.column))) {
dimFields += (field)
- }
- else if (isDetectAsDimentionDatatype(field.dataType.get)) {
+ } else if (isDetectAsDimentionDatatype(field.dataType.get)) {
dimFields += (field)
}
}
@@ -806,11 +823,12 @@ class CarbonSqlParser()
(dimFields.toSeq, noDictionaryDims)
}
+
/**
* It fills non string dimensions in dimFields
*/
def fillNonStringDimension(dictIncludeCols: Seq[String],
- field: Field, dimFields: LinkedHashSet[Field]) {
+ field: Field, dimFields: LinkedHashSet[Field]) {
var dictInclude = false
if (dictIncludeCols.nonEmpty) {
dictIncludeCols.foreach(dictIncludeCol =>
@@ -841,9 +859,9 @@ class CarbonSqlParser()
dimensionType.exists(x => x.equalsIgnoreCase(dimensionDataType))
}
- /**
- * detects whether double or decimal column is part of dictionary_exclude
- */
+ /**
+ * detects whether double or decimal column is part of dictionary_exclude
+ */
def isStringAndTimestampColDictionaryExclude(columnDataType: String): Boolean = {
val dataTypes = Array("string", "timestamp")
dataTypes.exists(x => x.equalsIgnoreCase(columnDataType))
@@ -857,7 +875,7 @@ class CarbonSqlParser()
* @return
*/
protected def extractMsrColsFromFields(fields: Seq[Field],
- tableProperties: Map[String, String]): Seq[Field] = {
+ tableProperties: Map[String, String]): Seq[Field] = {
var msrFields: Seq[Field] = Seq[Field]()
var dictIncludedCols: Array[String] = Array[String]()
var dictExcludedCols: Array[String] = Array[String]()
@@ -877,10 +895,10 @@ class CarbonSqlParser()
// by default consider all non string cols as msrs. consider all include/ exclude cols as dims
fields.foreach(field => {
if (!isDetectAsDimentionDatatype(field.dataType.get)) {
- if (!dictIncludedCols.exists(x => x.equalsIgnoreCase(field.column)) &&
+ if (!dictIncludedCols.exists(x => x.equalsIgnoreCase(field.column)) &&
!dictExcludedCols.exists(x => x.equalsIgnoreCase(field.column))) {
- msrFields :+= field
- }
+ msrFields :+= field
+ }
}
})
@@ -905,35 +923,43 @@ class CarbonSqlParser()
(db, tableName)
}
- protected def cleanIdentifier(ident: String): String = ident match {
- case escapedIdentifier(i) => i
- case plainIdent => plainIdent
+ protected def cleanIdentifier(ident: String): String = {
+ ident match {
+ case escapedIdentifier(i) => i
+ case plainIdent => plainIdent
+ }
}
protected def getClauses(clauseNames: Seq[String], nodeList: Seq[ASTNode]): Seq[Option[Node]] = {
var remainingNodes = nodeList
val clauses = clauseNames.map { clauseName =>
val (matches, nonMatches) = remainingNodes.partition(_.getText.toUpperCase == clauseName)
- remainingNodes = nonMatches ++ (if (matches.nonEmpty) matches.tail else Nil)
+ remainingNodes = nonMatches ++ (if (matches.nonEmpty) {
+ matches.tail
+ } else {
+ Nil
+ })
matches.headOption
}
if (remainingNodes.nonEmpty) {
sys.error(
s"""Unhandled clauses:
- |You are likely trying to use an unsupported carbon feature."""".stripMargin)
+ |You are likely trying to use an unsupported carbon feature."""".stripMargin)
}
clauses
}
object Token {
/** @return matches of the form (tokenName, children). */
- def unapply(t: Any): Option[(String, Seq[ASTNode])] = t match {
- case t: ASTNode =>
- CurrentOrigin.setPosition(t.getLine, t.getCharPositionInLine)
- Some((t.getText,
- Option(t.getChildren).map(_.asScala.toList).getOrElse(Nil).asInstanceOf[Seq[ASTNode]]))
- case _ => None
+ def unapply(t: Any): Option[(String, Seq[ASTNode])] = {
+ t match {
+ case t: ASTNode =>
+ CurrentOrigin.setPosition(t.getLine, t.getCharPositionInLine)
+ Some((t.getText,
+ Option(t.getChildren).map(_.asScala.toList).getOrElse(Nil).asInstanceOf[Seq[ASTNode]]))
+ case _ => None
+ }
}
}
@@ -943,35 +969,39 @@ class CarbonSqlParser()
* @param node
* @return
*/
- protected def getProperties(node: Node): Seq[(String, String)] = node match {
- case Token("TOK_TABLEPROPLIST", list) =>
- list.map {
- case Token("TOK_TABLEPROPERTY", Token(key, Nil) :: Token(value, Nil) :: Nil) =>
- (unquoteString(key) -> unquoteString(value))
- }
+ protected def getProperties(node: Node): Seq[(String, String)] = {
+ node match {
+ case Token("TOK_TABLEPROPLIST", list) =>
+ list.map {
+ case Token("TOK_TABLEPROPERTY", Token(key, Nil) :: Token(value, Nil) :: Nil) =>
+ (unquoteString(key) -> unquoteString(value))
+ }
+ }
}
- protected def unquoteString(str: String) = str match {
- case singleQuotedString(s) => s.toLowerCase()
- case doubleQuotedString(s) => s.toLowerCase()
- case other => other
+ protected def unquoteString(str: String) = {
+ str match {
+ case singleQuotedString(s) => s.toLowerCase()
+ case doubleQuotedString(s) => s.toLowerCase()
+ case other => other
+ }
}
protected lazy val loadDataNew: Parser[LogicalPlan] =
LOAD ~> DATA ~> opt(LOCAL) ~> INPATH ~> stringLit ~ opt(OVERWRITE) ~
- (INTO ~> TABLE ~> (ident <~ ".").? ~ ident) ~
- (OPTIONS ~> "(" ~> repsep(loadOptions, ",") <~ ")").? <~ opt(";") ^^ {
- case filePath ~ isOverwrite ~ table ~ optionsList =>
- val (databaseNameOp, tableName) = table match {
- case databaseName ~ tableName => (databaseName, tableName.toLowerCase())
- }
- if(optionsList.isDefined) {
- validateOptions(optionsList)
- }
- val optionsMap = optionsList.getOrElse(List.empty[(String, String)]).toMap
- LoadTable(databaseNameOp, tableName, filePath, Seq(), optionsMap,
- isOverwrite.isDefined)
- }
+ (INTO ~> TABLE ~> (ident <~ ".").? ~ ident) ~
+ (OPTIONS ~> "(" ~> repsep(loadOptions, ",") <~ ")").? <~ opt(";") ^^ {
+ case filePath ~ isOverwrite ~ table ~ optionsList =>
+ val (databaseNameOp, tableName) = table match {
+ case databaseName ~ tableName => (databaseName, tableName.toLowerCase())
+ }
+ if (optionsList.isDefined) {
+ validateOptions(optionsList)
+ }
+ val optionsMap = optionsList.getOrElse(List.empty[(String, String)]).toMap
+ LoadTable(databaseNameOp, tableName, filePath, Seq(), optionsMap,
+ isOverwrite.isDefined)
+ }
private def validateOptions(optionList: Option[List[(String, String)]]): Unit = {
@@ -999,9 +1029,9 @@ class CarbonSqlParser()
// COLUMNDICT and ALL_DICTIONARY_PATH can not be used together.
if (options.exists(_._1.equalsIgnoreCase("COLUMNDICT")) &&
- options.exists(_._1.equalsIgnoreCase("ALL_DICTIONARY_PATH"))) {
+ options.exists(_._1.equalsIgnoreCase("ALL_DICTIONARY_PATH"))) {
val errorMessage = "Error: COLUMNDICT and ALL_DICTIONARY_PATH can not be used together" +
- " in options"
+ " in options"
throw new MalformedCarbonCommandException(errorMessage)
}
@@ -1052,20 +1082,21 @@ class CarbonSqlParser()
protected lazy val primitiveTypes =
STRING ^^^ "string" | INTEGER ^^^ "integer" | TIMESTAMP ^^^
- "timestamp" | NUMERIC ^^^ "numeric" | BIGINT ^^^ "bigint" |
- INT ^^^ "int" | DOUBLE ^^^ "double" | decimalType
+ "timestamp" | NUMERIC ^^^ "numeric" |
+ BIGINT ^^^ "bigint" |
+ INT ^^^ "int" | DOUBLE ^^^ "double" | decimalType
/**
* Matching the decimal(10,0) data type and returning the same.
*/
private lazy val decimalType =
- DECIMAL ~ ("(" ~> numericLit <~",") ~ (numericLit <~ ")") ^^ {
- case decimal ~ precision ~scale =>
- s"$decimal($precision, $scale)"
- }
+ DECIMAL ~ ("(" ~> numericLit <~ ",") ~ (numericLit <~ ")") ^^ {
+ case decimal ~ precision ~ scale =>
+ s"$decimal($precision, $scale)"
+ }
protected lazy val nestedType: Parser[Field] = structFieldType | arrayFieldType |
- primitiveFieldType
+ primitiveFieldType
protected lazy val anyFieldDef: Parser[Field] =
(ident | stringLit) ~ ((":").? ~> nestedType) ~ (IN ~> (ident | stringLit)).? ^^ {
@@ -1095,8 +1126,8 @@ class CarbonSqlParser()
protected lazy val measureCol: Parser[Field] =
(ident | stringLit) ~ (INTEGER ^^^ "integer" | NUMERIC ^^^ "numeric" |
- BIGINT ^^^ "bigint" | DECIMAL ^^^ "decimal").? ~
- (AS ~> (ident | stringLit)).? ~ (IN ~> (ident | stringLit)).? ^^ {
+ BIGINT ^^^ "bigint" | DECIMAL ^^^ "decimal").? ~
+ (AS ~> (ident | stringLit)).? ~ (IN ~> (ident | stringLit)).? ^^ {
case e1 ~ e2 ~ e3 ~ e4 => Field(e1, e2, e3, Some(null))
}
@@ -1112,11 +1143,11 @@ class CarbonSqlParser()
if (ef.isDefined && "FORMATTED".equalsIgnoreCase(ef.get)) {
new DescribeFormattedCommand("describe formatted " + tblIdentifier,
tblIdentifier)
- }
- else {
+ } else {
new DescribeCommand(UnresolvedRelation(tblIdentifier, None), ef.isDefined)
}
}
+
private def normalizeType(field: Field): Field = {
val dataType = field.dataType.getOrElse("NIL")
dataType match {
@@ -1217,22 +1248,23 @@ class CarbonSqlParser()
protected lazy val showLoads: Parser[LogicalPlan] =
SHOW ~> SEGMENTS ~> FOR ~> TABLE ~> (ident <~ ".").? ~ ident ~
- (LIMIT ~> numericLit).? <~
- opt(";") ^^ {
+ (LIMIT ~> numericLit).? <~
+ opt(";") ^^ {
case databaseName ~ tableName ~ limit =>
ShowLoadsCommand(databaseName, tableName.toLowerCase(), limit)
}
protected lazy val segmentId: Parser[String] =
numericLit ^^ { u => u } |
- elem("decimal", p => {
- p.getClass.getSimpleName.equals("FloatLit") ||
- p.getClass.getSimpleName.equals("DecimalLit") } ) ^^ (_.chars)
+ elem("decimal", p => {
+ p.getClass.getSimpleName.equals("FloatLit") ||
+ p.getClass.getSimpleName.equals("DecimalLit")
+ }) ^^ (_.chars)
protected lazy val deleteLoadsByID: Parser[LogicalPlan] =
DELETE ~> SEGMENT ~> repsep(segmentId, ",") ~ (FROM ~> TABLE ~>
- (ident <~ ".").? ~ ident) <~
- opt(";") ^^ {
+ (ident <~ ".").? ~ ident) <~
+ opt(";") ^^ {
case loadids ~ table => table match {
case databaseName ~ tableName =>
DeleteLoadsById(loadids, databaseName, tableName.toLowerCase())
@@ -1241,8 +1273,8 @@ class CarbonSqlParser()
protected lazy val deleteLoadsByLoadDate: Parser[LogicalPlan] =
DELETE ~> SEGMENTS ~> FROM ~> TABLE ~> (ident <~ ".").? ~ ident ~
- (WHERE ~> (STARTTIME <~ BEFORE) ~ stringLit) <~
- opt(";") ^^ {
+ (WHERE ~> (STARTTIME <~ BEFORE) ~ stringLit) <~
+ opt(";") ^^ {
case schema ~ table ~ condition =>
condition match {
case dateField ~ dateValue =>
@@ -1261,6 +1293,7 @@ class CarbonSqlParser()
logicalPlan match {
case plan: CreateTable => ExplainCommand(logicalPlan, extended = isExtended.isDefined)
case _ => ExplainCommand(OneRowRelation)
- }
+ }
}
+
}