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 2017/08/17 01:29:18 UTC
[1/3] carbondata git commit: [CARBONDATA-940] alter table add/split
partition for spark 2.1
Repository: carbondata
Updated Branches:
refs/heads/master 3efd33072 -> 874764f95
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
index c059030..544a26a 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
@@ -504,6 +504,8 @@ public class CarbonFactDataHandlerModel {
return bucketId;
}
+ public void setBucketId(Integer bucketId) { this.bucketId = bucketId; }
+
public long getSchemaUpdatedTimeStamp() {
return schemaUpdatedTimeStamp;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
index f823ade..45461e3 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/util/CarbonDataProcessorUtil.java
@@ -181,9 +181,11 @@ public final class CarbonDataProcessorUtil {
* @return
*/
public static String[] getLocalDataFolderLocation(String databaseName, String tableName,
- String taskId, String partitionId, String segmentId, boolean isCompactionFlow) {
+ String taskId, String partitionId, String segmentId, boolean isCompactionFlow,
+ boolean isAltPartitionFlow) {
String tempLocationKey =
- getTempStoreLocationKey(databaseName, tableName, segmentId, taskId, isCompactionFlow);
+ getTempStoreLocationKey(databaseName, tableName, segmentId, taskId, isCompactionFlow,
+ isAltPartitionFlow);
String baseTempStorePath = CarbonProperties.getInstance()
.getProperty(tempLocationKey);
if (baseTempStorePath == null) {
@@ -221,13 +223,17 @@ public final class CarbonDataProcessorUtil {
* @return
*/
public static String getTempStoreLocationKey(String databaseName, String tableName,
- String segmentId, String taskId, boolean isCompactionFlow) {
+ String segmentId, String taskId, boolean isCompactionFlow, boolean isAltPartitionFlow) {
String tempLocationKey = databaseName + CarbonCommonConstants.UNDERSCORE + tableName
+ CarbonCommonConstants.UNDERSCORE + segmentId + CarbonCommonConstants.UNDERSCORE + taskId;
if (isCompactionFlow) {
tempLocationKey = CarbonCommonConstants.COMPACTION_KEY_WORD + CarbonCommonConstants.UNDERSCORE
+ tempLocationKey;
}
+ if (isAltPartitionFlow) {
+ tempLocationKey = CarbonCommonConstants.ALTER_PARTITION_KEY_WORD +
+ CarbonCommonConstants.UNDERSCORE + tempLocationKey;
+ }
return tempLocationKey;
}
[2/3] carbondata git commit: [CARBONDATA-940] alter table add/split
partition for spark 2.1
Posted by ja...@apache.org.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
index 3949404..184ab9e 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/util/PartitionUtils.scala
@@ -16,8 +16,28 @@
*/
package org.apache.spark.util
+import java.io.{File, IOException}
+import java.text.SimpleDateFormat
+import java.util
+
+import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hadoop.mapreduce.Job
+
+import org.apache.carbondata.core.datastore.block.{SegmentProperties, TableBlockInfo}
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
+import org.apache.carbondata.core.metadata.schema.PartitionInfo
+import org.apache.carbondata.core.metadata.schema.partition.PartitionType
+import org.apache.carbondata.core.util.CarbonUtil
+import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.hadoop.CarbonInputSplit
+import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
+import org.apache.carbondata.processing.model.CarbonLoadModel
+import org.apache.carbondata.spark.util.CommonUtil
+
object PartitionUtils {
def getListInfo(originListInfo: String): List[List[String]] = {
@@ -49,4 +69,119 @@ object PartitionUtils {
}
listInfo.toList
}
+
+ /**
+ * verify the add/split information and update the partitionInfo:
+ * 1. update rangeInfo/listInfo
+ * 2. update partitionIds
+ */
+ def updatePartitionInfo(partitionInfo: PartitionInfo, partitionIdList: List[Int],
+ partitionId: Int, splitInfo: List[String], timestampFormatter: SimpleDateFormat,
+ dateFormatter: SimpleDateFormat): Unit = {
+ val columnDataType = partitionInfo.getColumnSchemaList.get(0).getDataType
+ val index = partitionIdList.indexOf(partitionId)
+ if (partitionInfo.getPartitionType == PartitionType.RANGE) {
+ val rangeInfo = partitionInfo.getRangeInfo.asScala.toList
+ val newRangeInfo = partitionId match {
+ case 0 => rangeInfo ++ splitInfo
+ case _ => rangeInfo.take(index - 1) ++ splitInfo ++
+ rangeInfo.takeRight(rangeInfo.size - index)
+ }
+ CommonUtil.validateRangeInfo(newRangeInfo, columnDataType,
+ timestampFormatter, dateFormatter)
+ partitionInfo.setRangeInfo(newRangeInfo.asJava)
+ } else if (partitionInfo.getPartitionType == PartitionType.LIST) {
+ val originList = partitionInfo.getListInfo.asScala.map(_.asScala.toList).toList
+ if (partitionId != 0) {
+ val targetListInfo = partitionInfo.getListInfo.get(index - 1)
+ CommonUtil.validateSplitListInfo(targetListInfo.asScala.toList, splitInfo, originList)
+ } else {
+ CommonUtil.validateAddListInfo(splitInfo, originList)
+ }
+ val addListInfo = PartitionUtils.getListInfo(splitInfo.mkString(","))
+ val newListInfo = partitionId match {
+ case 0 => originList ++ addListInfo
+ case _ => originList.take(index - 1) ++ addListInfo ++
+ originList.takeRight(originList.size - index)
+ }
+ partitionInfo.setListInfo(newListInfo.map(_.asJava).asJava)
+ }
+
+ if (partitionId == 0) {
+ partitionInfo.addPartition(splitInfo.size)
+ } else {
+ partitionInfo.splitPartition(index, splitInfo.size)
+ }
+ }
+
+ /**
+ * Used for alter table partition commands to get segmentProperties in spark node
+ * @param identifier
+ * @param segmentId
+ * @param oldPartitionIdList Task id group before partition info is changed
+ * @return
+ */
+ def getSegmentProperties(identifier: AbsoluteTableIdentifier, segmentId: String,
+ partitionIds: List[String], oldPartitionIdList: List[Int],
+ partitionInfo: PartitionInfo): SegmentProperties = {
+ val tableBlockInfoList =
+ getPartitionBlockList(identifier, segmentId, partitionIds, oldPartitionIdList, partitionInfo)
+ val footer = CarbonUtil.readMetadatFile(tableBlockInfoList.get(0))
+ val segmentProperties = new SegmentProperties(footer.getColumnInTable,
+ footer.getSegmentInfo.getColumnCardinality)
+ segmentProperties
+ }
+
+ def getPartitionBlockList(identifier: AbsoluteTableIdentifier, segmentId: String,
+ partitionIds: List[String], oldPartitionIdList: List[Int],
+ partitionInfo: PartitionInfo): java.util.List[TableBlockInfo] = {
+ val jobConf = new JobConf(new Configuration)
+ val job = new Job(jobConf)
+ val format = CarbonInputFormatUtil
+ .createCarbonTableInputFormat(identifier, partitionIds.asJava, job)
+ val splits = format.getSplitsOfOneSegment(job, segmentId,
+ oldPartitionIdList.map(_.asInstanceOf[Integer]).asJava, partitionInfo)
+ val blockList = splits.asScala.map(_.asInstanceOf[CarbonInputSplit])
+ val tableBlockInfoList = CarbonInputSplit.createBlocks(blockList.asJava)
+ tableBlockInfoList
+ }
+
+ @throws(classOf[IOException])
+ def deleteOriginalCarbonFile(identifier: AbsoluteTableIdentifier, segmentId: String,
+ partitionIds: List[String], oldPartitionIdList: List[Int], storePath: String,
+ dbName: String, tableName: String, partitionInfo: PartitionInfo,
+ carbonLoadModel: CarbonLoadModel): Unit = {
+ val newTime = carbonLoadModel.getFactTimeStamp
+ val tableBlockInfoList =
+ getPartitionBlockList(identifier, segmentId, partitionIds, oldPartitionIdList,
+ partitionInfo).asScala
+ val pathList: util.List[String] = new util.ArrayList[String]()
+ val carbonTablePath = new CarbonTablePath(storePath, dbName, tableName)
+ tableBlockInfoList.foreach{ tableBlockInfo =>
+ val path = tableBlockInfo.getFilePath
+ val timestamp = CarbonTablePath.DataFileUtil.getTimeStampFromFileName(path)
+ if (timestamp.toLong != newTime) {
+ // add carbondata file
+ pathList.add(path)
+ // add index file
+ val version = tableBlockInfo.getVersion
+ val taskNo = CarbonTablePath.DataFileUtil.getTaskNo(path)
+ val batchNo = CarbonTablePath.DataFileUtil.getBatchNoFromTaskNo(taskNo)
+ val taskId = CarbonTablePath.DataFileUtil.getTaskIdFromTaskNo(taskNo)
+ val bucketNumber = CarbonTablePath.DataFileUtil.getBucketNo(path)
+ val indexFilePath = carbonTablePath.getCarbonIndexFilePath(String.valueOf(taskId), "0",
+ segmentId, batchNo, String.valueOf(bucketNumber), timestamp, version)
+ // indexFilePath could be duplicated when multiple data file related to one index file
+ if (indexFilePath != null && !pathList.contains(indexFilePath)) {
+ pathList.add(indexFilePath)
+ }
+ }
+ }
+ val files: util.List[File] = new util.ArrayList[File]()
+ for (path <- pathList.asScala) {
+ val file = new File(path)
+ files.add(file)
+ }
+ CarbonUtil.deleteFiles(files.asScala.toArray)
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
index 5ede835..f556a05 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -35,7 +35,7 @@ import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, FileSplit}
import org.apache.spark.{SparkEnv, SparkException}
import org.apache.spark.rdd.{DataLoadCoalescedRDD, DataLoadPartitionCoalescer, NewHadoopRDD, RDD, UpdateCoalescedRDD}
import org.apache.spark.sql.{CarbonEnv, DataFrame, Row, SQLContext}
-import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, ExecutionErrors, UpdateTableModel}
+import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.hive.DistributionUtil
import org.apache.spark.util.SparkUtil
@@ -47,23 +47,23 @@ import org.apache.carbondata.core.dictionary.server.DictionaryServer
import org.apache.carbondata.core.locks.{CarbonLockFactory, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.{CarbonTableIdentifier, ColumnarFormatVersion}
import org.apache.carbondata.core.metadata.datatype.DataType
+import org.apache.carbondata.core.metadata.schema.PartitionInfo
import org.apache.carbondata.core.metadata.schema.partition.PartitionType
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
import org.apache.carbondata.core.scan.partition.PartitionUtil
-import org.apache.carbondata.core.statusmanager.LoadMetadataDetails
+import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatusManager}
import org.apache.carbondata.core.util.{ByteUtil, CarbonProperties}
import org.apache.carbondata.core.util.path.CarbonStorePath
import org.apache.carbondata.processing.csvload.{BlockDetails, CSVInputFormat, StringArrayWritable}
import org.apache.carbondata.processing.etl.DataLoadingException
import org.apache.carbondata.processing.merger.{CarbonCompactionUtil, CarbonDataMergerUtil, CompactionType}
import org.apache.carbondata.processing.model.CarbonLoadModel
-import org.apache.carbondata.processing.newflow.exception.{BadRecordFoundException, CarbonDataLoadingException}
import org.apache.carbondata.processing.newflow.DataLoadProcessBuilder
-import org.apache.carbondata.processing.newflow.exception.CarbonDataLoadingException
+import org.apache.carbondata.processing.newflow.exception.{BadRecordFoundException, CarbonDataLoadingException}
import org.apache.carbondata.processing.newflow.sort.SortScopeOptions
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil
-import org.apache.carbondata.spark._
+import org.apache.carbondata.spark.{DataLoadResultImpl, PartitionFactory, _}
import org.apache.carbondata.spark.load.{FailureCauses, _}
import org.apache.carbondata.spark.splits.TableSplit
import org.apache.carbondata.spark.util.{CarbonQueryUtil, CarbonScalaUtil, CommonUtil}
@@ -178,6 +178,26 @@ object CarbonDataRDDFactory {
}
}
+ def alterTableSplitPartition(sqlContext: SQLContext,
+ partitionId: String,
+ carbonLoadModel: CarbonLoadModel,
+ storePath: String,
+ oldPartitionIdList: List[Int]): Unit = {
+ LOGGER.audit(s"Add partition request received for table " +
+ s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ try {
+ startSplitThreads(sqlContext,
+ carbonLoadModel,
+ storePath,
+ partitionId,
+ oldPartitionIdList)
+ } catch {
+ case e: Exception =>
+ LOGGER.error(s"Exception in start splitting partition thread. ${ e.getMessage }")
+ throw e
+ }
+ }
+
def handleCompactionForSystemLocking(sqlContext: SQLContext,
carbonLoadModel: CarbonLoadModel,
storePath: String,
@@ -345,6 +365,73 @@ object CarbonDataRDDFactory {
compactionThread.run()
}
+ case class SplitThread(sqlContext: SQLContext,
+ carbonLoadModel: CarbonLoadModel,
+ executor: ExecutorService,
+ storePath: String,
+ segmentId: String,
+ partitionId: String,
+ oldPartitionIdList: List[Int]) extends Thread {
+ override def run(): Unit = {
+ var triggeredSplitPartitionStatus = false
+ var exception: Exception = null
+ try {
+ DataManagementFunc.executePartitionSplit(sqlContext,
+ carbonLoadModel, executor, storePath, segmentId, partitionId,
+ oldPartitionIdList)
+ triggeredSplitPartitionStatus = true
+ } catch {
+ case e: Exception =>
+ LOGGER.error(s"Exception in partition split thread: ${ e.getMessage } }")
+ exception = e
+ }
+ if (triggeredSplitPartitionStatus == false) {
+ throw new Exception("Exception in split partition " + exception.getMessage)
+ }
+ }
+ }
+
+ def startSplitThreads(sqlContext: SQLContext,
+ carbonLoadModel: CarbonLoadModel,
+ storePath: String,
+ partitionId: String,
+ oldPartitionIdList: List[Int]): Unit = {
+ val numberOfCores = CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.NUM_CORES_ALT_PARTITION,
+ CarbonCommonConstants.DEFAULT_NUMBER_CORES)
+ val executor : ExecutorService = Executors.newFixedThreadPool(numberOfCores.toInt)
+ try {
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val absoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
+ val segmentStatusManager = new SegmentStatusManager(absoluteTableIdentifier)
+ val validSegments = segmentStatusManager.getValidAndInvalidSegments.getValidSegments.asScala
+ val threadArray: Array[SplitThread] = new Array[SplitThread](validSegments.size)
+ var i = 0
+ validSegments.foreach { segmentId =>
+ threadArray(i) = SplitThread(sqlContext, carbonLoadModel, executor, storePath,
+ segmentId, partitionId, oldPartitionIdList)
+ threadArray(i).start()
+ i += 1
+ }
+ threadArray.foreach {
+ thread => thread.join()
+ }
+ } catch {
+ case e: Exception =>
+ LOGGER.error(s"Exception when split partition: ${ e.getMessage }")
+ throw e
+ } finally {
+ executor.shutdown()
+ try {
+ CarbonLoaderUtil.deletePartialLoadDataIfExist(carbonLoadModel, false)
+ } catch {
+ case e: Exception =>
+ LOGGER.error(s"Exception in add/split partition thread while deleting partial load file" +
+ s" ${ e.getMessage }")
+ }
+ }
+ }
+
def loadCarbonData(sqlContext: SQLContext,
carbonLoadModel: CarbonLoadModel,
storePath: String,
@@ -946,7 +1033,7 @@ object CarbonDataRDDFactory {
}
/**
- * repartition the input data for partiton table.
+ * repartition the input data for partition table.
* @param sqlContext
* @param dataFrame
* @param carbonLoadModel
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index 4711618..9cc12cd 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -17,6 +17,9 @@
package org.apache.spark.sql.execution.command
+import java.text.SimpleDateFormat
+import java.util
+
import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
import scala.language.implicitConversions
@@ -29,20 +32,24 @@ import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.expressions.Attribute
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.SparkPlan
-import org.apache.spark.sql.hive.CarbonRelation
-import org.apache.spark.util.FileUtils
+import org.apache.spark.sql.hive.{CarbonMetaStore, CarbonRelation}
+import org.apache.spark.util.{AlterTableUtil, FileUtils, PartitionUtils}
import org.codehaus.jackson.map.ObjectMapper
import org.apache.carbondata.api.CarbonStore
import org.apache.carbondata.common.constants.LoggerAction
import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.cache.CacheProvider
import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.dictionary.server.DictionaryServer
-import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil, ICarbonLock, LockUsage}
-import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
+import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage}
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonMetadata, CarbonTableIdentifier}
+import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
import org.apache.carbondata.core.metadata.encoder.Encoding
-import org.apache.carbondata.core.metadata.schema.table.TableInfo
+import org.apache.carbondata.core.metadata.schema.PartitionInfo
+import org.apache.carbondata.core.metadata.schema.partition.PartitionType
+import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo}
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
import org.apache.carbondata.core.mutate.{CarbonUpdateUtil, TupleIdEnum}
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
@@ -183,6 +190,136 @@ case class AlterTableCompaction(alterTableModel: AlterTableModel) extends Runnab
}
}
+/**
+ * Command for Alter Table Add & Split partition
+ * Add is a special case of Splitting the default partition (part0)
+ * @param splitPartitionModel
+ */
+case class AlterTableSplitPartitionCommand(splitPartitionModel: AlterTableSplitPartitionModel)
+ extends RunnableCommand with DataProcessCommand with SchemaProcessCommand {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+ val tableName = splitPartitionModel.tableName
+ val splitInfo = splitPartitionModel.splitInfo
+ val partitionId = splitPartitionModel.partitionId.toInt
+ var partitionInfo: PartitionInfo = null
+ var carbonMetaStore: CarbonMetaStore = null
+ var relation: CarbonRelation = null
+ var dbName: String = null
+ var storePath: String = null
+ var table: CarbonTable = null
+ var carbonTableIdentifier: CarbonTableIdentifier = null
+ val oldPartitionIds: util.ArrayList[Int] = new util.ArrayList[Int]()
+ val timestampFormatter = new SimpleDateFormat(CarbonProperties.getInstance
+ .getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
+ CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT))
+ val dateFormatter = new SimpleDateFormat(CarbonProperties.getInstance
+ .getProperty(CarbonCommonConstants.CARBON_DATE_FORMAT,
+ CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT))
+ val locksToBeAcquired = List(LockUsage.METADATA_LOCK,
+ LockUsage.COMPACTION_LOCK,
+ LockUsage.DELETE_SEGMENT_LOCK,
+ LockUsage.DROP_TABLE_LOCK,
+ LockUsage.CLEAN_FILES_LOCK,
+ LockUsage.ALTER_PARTITION_LOCK)
+
+ // TODO will add rollback function incase process data failure
+ def run(sparkSession: SparkSession): Seq[Row] = {
+ processSchema(sparkSession)
+ processData(sparkSession)
+ }
+
+ override def processSchema(sparkSession: SparkSession): Seq[Row] = {
+ dbName = splitPartitionModel.databaseName
+ .getOrElse(sparkSession.catalog.currentDatabase)
+ carbonMetaStore = CarbonEnv.getInstance(sparkSession).carbonMetastore
+ relation = carbonMetaStore.lookupRelation(Option(dbName), tableName)(sparkSession)
+ .asInstanceOf[CarbonRelation]
+ carbonTableIdentifier = relation.tableMeta.carbonTableIdentifier
+ storePath = relation.tableMeta.storePath
+ if (relation == null) {
+ sys.error(s"Table $dbName.$tableName does not exist")
+ }
+ carbonMetaStore.checkSchemasModifiedTimeAndReloadTables(storePath)
+ if (null == CarbonMetadata.getInstance.getCarbonTable(dbName + "_" + tableName)) {
+ LOGGER.error(s"Alter table failed. table not found: $dbName.$tableName")
+ sys.error(s"Alter table failed. table not found: $dbName.$tableName")
+ }
+ table = relation.tableMeta.carbonTable
+ partitionInfo = table.getPartitionInfo(tableName)
+ val partitionIds = partitionInfo.getPartitionIds.asScala.map(_.asInstanceOf[Int]).toList
+ // keep a copy of partitionIdList before update partitionInfo.
+ // will be used in partition data scan
+ oldPartitionIds.addAll(partitionIds.asJava)
+
+ if (partitionInfo == null) {
+ sys.error(s"Table $tableName is not a partition table.")
+ }
+ if (partitionInfo.getPartitionType == PartitionType.HASH) {
+ sys.error(s"Hash partition table cannot be added or split!")
+ }
+ PartitionUtils.updatePartitionInfo(partitionInfo, partitionIds, partitionId,
+ splitInfo, timestampFormatter, dateFormatter)
+
+ val carbonTablePath = CarbonStorePath.getCarbonTablePath(storePath, carbonTableIdentifier)
+ val schemaFilePath = carbonTablePath.getSchemaFilePath
+ // read TableInfo
+ val tableInfo = carbonMetaStore.getThriftTableInfo(carbonTablePath)(sparkSession)
+ val schemaConverter = new ThriftWrapperSchemaConverterImpl()
+ val wrapperTableInfo = schemaConverter.fromExternalToWrapperTableInfo(tableInfo,
+ dbName, tableName, storePath)
+ val tableSchema = wrapperTableInfo.getFactTable
+ tableSchema.setPartitionInfo(partitionInfo)
+ wrapperTableInfo.setFactTable(tableSchema)
+ wrapperTableInfo.setLastUpdatedTime(System.currentTimeMillis())
+ val thriftTable =
+ schemaConverter.fromWrapperToExternalTableInfo(wrapperTableInfo, dbName, tableName)
+ carbonMetaStore.updateMetadataByThriftTable(schemaFilePath, thriftTable,
+ dbName, tableName, storePath)
+ CarbonUtil.writeThriftTableToSchemaFile(schemaFilePath, thriftTable)
+ // update the schema modified time
+ carbonMetaStore.updateAndTouchSchemasUpdatedTime(storePath)
+ sparkSession.catalog.refreshTable(tableName)
+ Seq.empty
+ }
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ var locks = List.empty[ICarbonLock]
+ var success = false
+ try {
+ locks = AlterTableUtil.validateTableAndAcquireLock(dbName, tableName,
+ locksToBeAcquired)(sparkSession)
+ val carbonLoadModel = new CarbonLoadModel()
+ val dataLoadSchema = new CarbonDataLoadSchema(table)
+ carbonLoadModel.setCarbonDataLoadSchema(dataLoadSchema)
+ carbonLoadModel.setTableName(carbonTableIdentifier.getTableName)
+ carbonLoadModel.setDatabaseName(carbonTableIdentifier.getDatabaseName)
+ carbonLoadModel.setStorePath(storePath)
+ val loadStartTime = CarbonUpdateUtil.readCurrentTime
+ carbonLoadModel.setFactTimeStamp(loadStartTime)
+ CarbonDataRDDFactory.alterTableSplitPartition(sparkSession.sqlContext,
+ partitionId.toString,
+ carbonLoadModel,
+ relation.tableMeta.storePath,
+ oldPartitionIds.asScala.toList
+ )
+ success = true
+ } catch {
+ case e: Exception =>
+ success = false
+ sys.error(s"Add/Split Partition failed. Please check logs for more info. ${ e.getMessage }")
+ } finally {
+ AlterTableUtil.releaseLocks(locks)
+ CacheProvider.getInstance().dropAllCache()
+ LOGGER.info("Locks released after alter table add/split partition action.")
+ if (success) {
+ LOGGER.info(s"Alter table add/split partition is successful for table $dbName.$tableName")
+ LOGGER.audit(s"Alter table add/split partition is successful for table $dbName.$tableName")
+ }
+ }
+ Seq.empty
+ }
+}
+
case class CreateTable(cm: TableModel, createDSTable: Boolean = true) extends RunnableCommand
with SchemaProcessCommand {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index 511a61c..780839a 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -63,7 +63,7 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
protected lazy val start: Parser[LogicalPlan] = explainPlan | startCommand
protected lazy val startCommand: Parser[LogicalPlan] =
- loadManagement| showLoads | alterTable | restructure | updateTable | deleteRecords
+ loadManagement|showLoads|alterTable|restructure|updateTable|deleteRecords|alterPartition
protected lazy val loadManagement: Parser[LogicalPlan] =
deleteLoadsByID | deleteLoadsByLoadDate | cleanFiles | loadDataNew
@@ -71,6 +71,30 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
protected lazy val restructure: Parser[LogicalPlan] =
alterTableModifyDataType | alterTableDropColumn | alterTableAddColumns
+ protected lazy val alterPartition: Parser[LogicalPlan] =
+ alterAddPartition | alterSplitPartition
+
+ protected lazy val alterAddPartition: Parser[LogicalPlan] =
+ ALTER ~> TABLE ~> (ident <~ ".").? ~ ident ~ (ADD ~> PARTITION ~>
+ "(" ~> repsep(stringLit, ",") <~ ")") <~ opt(";") ^^ {
+ case dbName ~ table ~ addInfo =>
+ val alterTableAddPartitionModel =
+ AlterTableSplitPartitionModel(dbName, table, "0", addInfo)
+ AlterTableSplitPartitionCommand(alterTableAddPartitionModel)
+ }
+
+ protected lazy val alterSplitPartition: Parser[LogicalPlan] =
+ ALTER ~> TABLE ~> (ident <~ ".").? ~ ident ~ (SPLIT ~> PARTITION ~>
+ "(" ~> numericLit <~ ")") ~ (INTO ~> "(" ~> repsep(stringLit, ",") <~ ")") <~ opt(";") ^^ {
+ case dbName ~ table ~ partitionId ~ splitInfo =>
+ val alterTableSplitPartitionModel =
+ AlterTableSplitPartitionModel(dbName, table, partitionId, splitInfo)
+ if (partitionId == 0) {
+ sys.error("Please use [Alter Table Add Partition] statement to split default partition!")
+ }
+ AlterTableSplitPartitionCommand(alterTableSplitPartitionModel)
+ }
+
protected lazy val alterTable: Parser[LogicalPlan] =
ALTER ~> TABLE ~> (ident <~ ".").? ~ ident ~ (COMPACT ~ stringLit) <~ opt(";") ^^ {
case dbName ~ table ~ (compact ~ compactType) =>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
index 41d6bd3..b4389a6 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
@@ -123,7 +123,7 @@ class CarbonSqlAstBuilder(conf: SQLConf) extends SparkSqlAstBuilder(conf) {
// validate partition clause
if (partitionerFields.nonEmpty) {
if (!CommonUtil.validatePartitionColumns(tableProperties, partitionerFields)) {
- throw new MalformedCarbonCommandException("Invalid partition definition")
+ throw new MalformedCarbonCommandException("Error: Invalid partition definition")
}
// partition columns should not be part of the schema
val badPartCols = partitionerFields.map(_.partitionColumn).toSet.intersect(colNames.toSet)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
new file mode 100644
index 0000000..7d86468
--- /dev/null
+++ b/integration/spark2/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestAlterPartitionTable.scala
@@ -0,0 +1,523 @@
+/*
+ * 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.testsuite.partition
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ListBuffer
+
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
+import org.apache.carbondata.core.datastore.impl.FileFactory
+import org.apache.carbondata.core.metadata.CarbonMetadata
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.core.util.path.CarbonTablePath
+
+class TestAlterPartitionTable extends QueryTest with BeforeAndAfterAll {
+
+
+ override def beforeAll {
+ dropTable
+
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
+ /**
+ * list_table_area_origin
+ * list_table_area
+ */
+ sql("""
+ | CREATE TABLE IF NOT EXISTS list_table_area_origin
+ | (
+ | id Int,
+ | vin string,
+ | logdate Timestamp,
+ | phonenumber Long,
+ | country string,
+ | salary Int
+ | )
+ | PARTITIONED BY (area string)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='LIST',
+ | 'LIST_INFO'='Asia, America, Europe')
+ """.stripMargin)
+ sql("""
+ | CREATE TABLE IF NOT EXISTS list_table_area
+ | (
+ | id Int,
+ | vin string,
+ | logdate Timestamp,
+ | phonenumber Long,
+ | country string,
+ | salary Int
+ | )
+ | PARTITIONED BY (area string)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='LIST',
+ | 'LIST_INFO'='Asia, America, Europe')
+ """.stripMargin)
+
+ /**
+ * range_table_logdate_origin
+ * range_table_logdate
+ */
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS range_table_logdate_origin
+ | (
+ | id Int,
+ | vin string,
+ | phonenumber Long,
+ | country string,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (logdate Timestamp)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01')
+ """.stripMargin)
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS range_table_logdate
+ | (
+ | id Int,
+ | vin string,
+ | phonenumber Long,
+ | country string,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (logdate Timestamp)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01')
+ """.stripMargin)
+
+ /**
+ * list_table_country_origin
+ * list_table_country
+ */
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS list_table_country_origin
+ | (
+ | id Int,
+ | vin string,
+ | logdate Timestamp,
+ | phonenumber Long,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (country string)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='LIST',
+ | 'LIST_INFO'='(China, US),UK ,Japan,(Canada,Russia, Good, NotGood), Korea ')
+ """.stripMargin)
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS list_table_country
+ | (
+ | id Int,
+ | vin string,
+ | logdate Timestamp,
+ | phonenumber Long,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (country string)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='LIST',
+ | 'LIST_INFO'='(China, US),UK ,Japan,(Canada,Russia, Good, NotGood), Korea ')
+ """.stripMargin)
+
+ /**
+ * range_table_logdate_split_origin
+ * range_table_logdate_split
+ */
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS range_table_logdate_split_origin
+ | (
+ | id Int,
+ | vin string,
+ | phonenumber Long,
+ | country string,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (logdate Timestamp)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01, 2018/01/01')
+ """.stripMargin)
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS range_table_logdate_split
+ | (
+ | id Int,
+ | vin string,
+ | phonenumber Long,
+ | country string,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (logdate Timestamp)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01, 2018/01/01')
+ """.stripMargin)
+
+ /**
+ * range_table_bucket_origin
+ * range_table_bucket
+ */
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS range_table_bucket_origin
+ | (
+ | id Int,
+ | vin string,
+ | phonenumber Long,
+ | country string,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (logdate Timestamp)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01, 2018/01/01',
+ | 'BUCKETNUMBER'='3',
+ | 'BUCKETCOLUMNS'='country')
+ """.stripMargin)
+ sql(
+ """
+ | CREATE TABLE IF NOT EXISTS range_table_bucket
+ | (
+ | id Int,
+ | vin string,
+ | phonenumber Long,
+ | country string,
+ | area string,
+ | salary Int
+ | )
+ | PARTITIONED BY (logdate Timestamp)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01, 2018/01/01',
+ | 'BUCKETNUMBER'='3',
+ | 'BUCKETCOLUMNS'='country')
+ """.stripMargin)
+
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE list_table_area_origin OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE range_table_logdate_origin OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE list_table_country_origin OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE range_table_logdate_split_origin OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE range_table_bucket_origin OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE list_table_area OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE range_table_logdate OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE list_table_country OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE range_table_logdate_split OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+ sql(s"""LOAD DATA LOCAL INPATH '$resourcesPath/partition_data.csv' INTO TABLE range_table_bucket OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
+
+ }
+
+ test("Alter table add partition: List Partition") {
+ sql("""ALTER TABLE list_table_area ADD PARTITION ('OutSpace', 'Hi')""".stripMargin)
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable("default_list_table_area")
+ val partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName)
+ val partitionIds = partitionInfo.getPartitionIds
+ val list_info = partitionInfo.getListInfo
+ assert(partitionIds == List(0, 1, 2, 3, 4, 5).map(Integer.valueOf(_)).asJava)
+ assert(partitionInfo.getMAX_PARTITION == 5)
+ assert(partitionInfo.getNumPartitions == 6)
+ assert(list_info.get(0).get(0) == "Asia")
+ assert(list_info.get(1).get(0) == "America")
+ assert(list_info.get(2).get(0) == "Europe")
+ assert(list_info.get(3).get(0) == "OutSpace")
+ assert(list_info.get(4).get(0) == "Hi")
+ validateDataFiles("default_list_table_area", "0", Seq(0, 1, 2, 4))
+ val result_after = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area")
+ val result_origin = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin")
+ checkAnswer(result_after, result_origin)
+
+ val result_after1 = sql(s"select id, vin, logdate, phonenumber, country, area, salary from list_table_area where area < 'OutSpace' ")
+ val rssult_origin1 = sql(s"select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin where area < 'OutSpace' ")
+ checkAnswer(result_after1, rssult_origin1)
+
+ val result_after2 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area where area <= 'OutSpace' ")
+ val result_origin2 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin where area <= 'OutSpace' ")
+ checkAnswer(result_after2, result_origin2)
+
+ val result_after3 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area where area = 'OutSpace' ")
+ val result_origin3 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin where area = 'OutSpace' ")
+ checkAnswer(result_after3, result_origin3)
+
+ val result_after4 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area where area > 'OutSpace' ")
+ val result_origin4 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin where area > 'OutSpace' ")
+ checkAnswer(result_after4, result_origin4)
+
+ val result_after5 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area where area >= 'OutSpace' ")
+ val result_origin5 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin where area >= 'OutSpace' ")
+ checkAnswer(result_after5, result_origin5)
+
+ sql("""ALTER TABLE list_table_area ADD PARTITION ('One', '(Two, Three)', 'Four')""".stripMargin)
+ val carbonTable1 = CarbonMetadata.getInstance().getCarbonTable("default_list_table_area")
+ val partitionInfo1 = carbonTable1.getPartitionInfo(carbonTable.getFactTableName)
+ val partitionIds1 = partitionInfo1.getPartitionIds
+ val new_list_info = partitionInfo1.getListInfo
+ assert(partitionIds1 == List(0, 1, 2, 3, 4, 5, 6, 7, 8).map(Integer.valueOf(_)).asJava)
+ assert(partitionInfo1.getMAX_PARTITION == 8)
+ assert(partitionInfo1.getNumPartitions == 9)
+ assert(new_list_info.get(0).get(0) == "Asia")
+ assert(new_list_info.get(1).get(0) == "America")
+ assert(new_list_info.get(2).get(0) == "Europe")
+ assert(new_list_info.get(3).get(0) == "OutSpace")
+ assert(new_list_info.get(4).get(0) == "Hi")
+ assert(new_list_info.get(5).get(0) == "One")
+ assert(new_list_info.get(6).get(0) == "Two")
+ assert(new_list_info.get(6).get(1) == "Three")
+ assert(new_list_info.get(7).get(0) == "Four")
+ validateDataFiles("default_list_table_area", "0", Seq(0, 1, 2, 4))
+
+ val result_after6 = sql("select id, vin, logdate, phonenumber, country, area, salary from list_table_area")
+ val result_origin6 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_area_origin""")
+ checkAnswer(result_after6, result_origin6)
+ }
+
+ test("Alter table add partition: Range Partition") {
+ sql("""ALTER TABLE range_table_logdate ADD PARTITION ('2017/01/01', '2018/01/01')""")
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable("default_range_table_logdate")
+ val partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName)
+ val partitionIds = partitionInfo.getPartitionIds
+ val range_info = partitionInfo.getRangeInfo
+ assert(partitionIds.size() == 6)
+ assert(partitionIds == List(0, 1, 2, 3, 4, 5).map(Integer.valueOf(_)).asJava)
+ assert(partitionInfo.getMAX_PARTITION == 5)
+ assert(range_info.get(0) == "2014/01/01")
+ assert(range_info.get(1) == "2015/01/01")
+ assert(range_info.get(2) == "2016/01/01")
+ assert(range_info.get(3) == "2017/01/01")
+ assert(range_info.get(4) == "2018/01/01")
+ validateDataFiles("default_range_table_logdate", "0", Seq(1, 2, 3, 4, 5))
+ val result_after = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate""")
+ val result_origin = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_origin""")
+ checkAnswer(result_after, result_origin)
+
+ val result_after1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate where logdate < cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_origin where logdate < cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after1, result_origin1)
+
+ val result_after2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate where logdate <= cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_origin where logdate <= cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after2, result_origin2)
+
+ val result_after3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate where logdate = cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_origin where logdate = cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after3, result_origin3)
+
+ val result_after4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate where logdate >= cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_origin where logdate >= cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after4, result_origin4)
+
+ val result_after5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate where logdate > cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_origin where logdate > cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after5, result_origin5)
+ }
+
+ test("Alter table split partition: List Partition") {
+ sql("""ALTER TABLE list_table_country SPLIT PARTITION(4) INTO ('Canada', 'Russia', '(Good, NotGood)')""".stripMargin)
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable("default_list_table_country")
+ val partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName)
+ val partitionIds = partitionInfo.getPartitionIds
+ val list_info = partitionInfo.getListInfo
+ assert(partitionIds == List(0, 1, 2, 3, 6, 7, 8, 5).map(Integer.valueOf(_)).asJava)
+ assert(partitionInfo.getMAX_PARTITION == 8)
+ assert(partitionInfo.getNumPartitions == 8)
+ assert(list_info.get(0).get(0) == "China")
+ assert(list_info.get(0).get(1) == "US")
+ assert(list_info.get(1).get(0) == "UK")
+ assert(list_info.get(2).get(0) == "Japan")
+ assert(list_info.get(3).get(0) == "Canada")
+ assert(list_info.get(4).get(0) == "Russia")
+ assert(list_info.get(5).get(0) == "Good")
+ assert(list_info.get(5).get(1) == "NotGood")
+ assert(list_info.get(6).get(0) == "Korea")
+ validateDataFiles("default_list_table_country", "0", Seq(0, 1, 2, 3, 8))
+ val result_after = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country""")
+ val result_origin = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country_origin""")
+ checkAnswer(result_after, result_origin)
+
+ val result_after1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country where country < 'NotGood' """)
+ val result_origin1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country_origin where country < 'NotGood' """)
+ checkAnswer(result_after1, result_origin1)
+
+ val result_after2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country where country <= 'NotGood' """)
+ val result_origin2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country_origin where country <= 'NotGood' """)
+ checkAnswer(result_after2, result_origin2)
+
+ val result_after3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country where country = 'NotGood' """)
+ val result_origin3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country_origin where country = 'NotGood' """)
+ checkAnswer(result_after3, result_origin3)
+
+ val result_after4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country where country >= 'NotGood' """)
+ val result_origin4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country_origin where country >= 'NotGood' """)
+ checkAnswer(result_after4, result_origin4)
+
+ val result_after5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country where country > 'NotGood' """)
+ val result_origin5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from list_table_country_origin where country > 'NotGood' """)
+ checkAnswer(result_after5, result_origin5)
+ }
+
+ test("Alter table split partition: Range Partition") {
+ sql("""ALTER TABLE range_table_logdate_split SPLIT PARTITION(4) INTO ('2017/01/01', '2018/01/01')""")
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable("default_range_table_logdate_split")
+ val partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName)
+ val partitionIds = partitionInfo.getPartitionIds
+ val rangeInfo = partitionInfo.getRangeInfo
+ assert(partitionIds == List(0, 1, 2, 3, 5, 6).map(Integer.valueOf(_)).asJava)
+ assert(partitionInfo.getMAX_PARTITION == 6)
+ assert(partitionInfo.getNumPartitions == 6)
+ assert(rangeInfo.get(0) == "2014/01/01")
+ assert(rangeInfo.get(1) == "2015/01/01")
+ assert(rangeInfo.get(2) == "2016/01/01")
+ assert(rangeInfo.get(3) == "2017/01/01")
+ assert(rangeInfo.get(4) == "2018/01/01")
+ validateDataFiles("default_range_table_logdate_split", "0", Seq(1, 2, 3, 5, 6))
+ val result_after = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split""")
+ val result_origin = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split_origin""")
+ checkAnswer(result_after, result_origin)
+
+ val result_after1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split where logdate < cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split_origin where logdate < cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after1, result_origin1)
+
+ val result_after2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split where logdate <= cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split_origin where logdate <= cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after2, result_origin2)
+
+ val result_after3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split where logdate = cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split_origin where logdate = cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after3, result_origin3)
+
+ val result_after4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split where logdate >= cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split_origin where logdate >= cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after4, result_origin4)
+
+ val result_after5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split where logdate > cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_logdate_split_origin where logdate > cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after5, result_origin5)
+ }
+
+ test("Alter table split partition: Range Partition + Bucket") {
+ sql("""ALTER TABLE range_table_bucket SPLIT PARTITION(4) INTO ('2017/01/01', '2018/01/01')""")
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable("default_range_table_bucket")
+ val partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName)
+ val partitionIds = partitionInfo.getPartitionIds
+ val rangeInfo = partitionInfo.getRangeInfo
+ assert(partitionIds == List(0, 1, 2, 3, 5, 6).map(Integer.valueOf(_)).asJava)
+ assert(partitionInfo.getMAX_PARTITION == 6)
+ assert(partitionInfo.getNumPartitions == 6)
+ assert(rangeInfo.get(0) == "2014/01/01")
+ assert(rangeInfo.get(1) == "2015/01/01")
+ assert(rangeInfo.get(2) == "2016/01/01")
+ assert(rangeInfo.get(3) == "2017/01/01")
+ assert(rangeInfo.get(4) == "2018/01/01")
+ validateDataFiles("default_range_table_bucket", "0", Seq(1, 2, 3, 5, 6))
+ val result_after = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket""")
+ val result_origin = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket_origin""")
+ checkAnswer(result_after, result_origin)
+
+ val result_after1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket where logdate < cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin1 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket_origin where logdate < cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after1, result_origin1)
+
+ val result_after2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket where logdate <= cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin2 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket_origin where logdate <= cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after2, result_origin2)
+
+ val result_origin3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket where logdate = cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_after3 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket_origin where logdate = cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_origin3, result_after3)
+
+ val result_after4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket where logdate >= cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin4 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket_origin where logdate >= cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after4, result_origin4)
+
+ val result_after5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket where logdate > cast('2017/01/12 00:00:00' as timestamp) """)
+ val result_origin5 = sql("""select id, vin, logdate, phonenumber, country, area, salary from range_table_bucket_origin where logdate > cast('2017/01/12 00:00:00' as timestamp) """)
+ checkAnswer(result_after5, result_origin5)
+ }
+
+ def validateDataFiles(tableUniqueName: String, segmentId: String, partitions: Seq[Int]): Unit = {
+ val carbonTable = CarbonMetadata.getInstance().getCarbonTable(tableUniqueName)
+ val dataFiles = getDataFiles(carbonTable, segmentId)
+ validatePartitionTableFiles(partitions, dataFiles)
+ }
+
+ def getDataFiles(carbonTable: CarbonTable, segmentId: String): Array[CarbonFile] = {
+ val tablePath = new CarbonTablePath(carbonTable.getStorePath, carbonTable.getDatabaseName,
+ carbonTable.getFactTableName)
+ val segmentDir = tablePath.getCarbonDataDirectoryPath("0", segmentId)
+ val carbonFile = FileFactory.getCarbonFile(segmentDir, FileFactory.getFileType(segmentDir))
+ val dataFiles = carbonFile.listFiles(new CarbonFileFilter() {
+ override def accept(file: CarbonFile): Boolean = {
+ return file.getName.endsWith(".carbondata")
+ }
+ })
+ dataFiles
+ }
+
+ /**
+ * should ensure answer equals to expected list, not only contains
+ * @param partitions
+ * @param dataFiles
+ */
+ def validatePartitionTableFiles(partitions: Seq[Int], dataFiles: Array[CarbonFile]): Unit = {
+ val partitionIds: ListBuffer[Int] = new ListBuffer[Int]()
+ dataFiles.foreach { dataFile =>
+ val partitionId = CarbonTablePath.DataFileUtil.getTaskNo(dataFile.getName).split("_")(0).toInt
+ partitionIds += partitionId
+ assert(partitions.contains(partitionId))
+ }
+ partitions.foreach(id => assert(partitionIds.contains(id)))
+ }
+
+ override def afterAll = {
+ dropTable
+ CarbonProperties.getInstance()
+ .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
+ }
+
+ def dropTable {
+ sql("DROP TABLE IF EXISTS list_table_area_origin")
+ sql("DROP TABLE IF EXISTS range_table_logdate_origin")
+ sql("DROP TABLE IF EXISTS list_table_country_origin")
+ sql("DROP TABLE IF EXISTS range_table_logdate_split_origin")
+ sql("DROP TABLE IF EXISTS range_table_bucket_origin")
+ sql("DROP TABLE IF EXISTS list_table_area")
+ sql("DROP TABLE IF EXISTS range_table_logdate")
+ sql("DROP TABLE IF EXISTS list_table_country")
+ sql("DROP TABLE IF EXISTS range_table_logdate_split")
+ sql("DROP TABLE IF EXISTS range_table_bucket")
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
index 56aaf54..095e5a3 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/merger/CompactionResultSortProcessor.java
@@ -402,7 +402,8 @@ public class CompactionResultSortProcessor extends AbstractResultProcessor {
private void initTempStoreLocation() {
tempStoreLocation = CarbonDataProcessorUtil
.getLocalDataFolderLocation(carbonLoadModel.getDatabaseName(), tableName,
- carbonLoadModel.getTaskNo(), carbonLoadModel.getPartitionId(), segmentId, true);
+ carbonLoadModel.getTaskNo(), carbonLoadModel.getPartitionId(), segmentId,
+ true, false);
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
index 518d64b..ccb25e6 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/DataLoadProcessBuilder.java
@@ -143,7 +143,7 @@ public final class DataLoadProcessBuilder {
String tableName = loadModel.getTableName();
String tempLocationKey = CarbonDataProcessorUtil
.getTempStoreLocationKey(databaseName, tableName, loadModel.getSegmentId(),
- loadModel.getTaskNo(), false);
+ loadModel.getTaskNo(), false, false);
CarbonProperties.getInstance().addProperty(tempLocationKey,
StringUtils.join(storeLocation, File.pathSeparator));
CarbonProperties.getInstance()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterImpl.java
index 1aa06f6..c3cf3c0 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterImpl.java
@@ -72,7 +72,7 @@ public class ParallelReadMergeSorterImpl extends AbstractMergeSorter {
CarbonDataProcessorUtil.getLocalDataFolderLocation(
sortParameters.getDatabaseName(), sortParameters.getTableName(),
String.valueOf(sortParameters.getTaskNo()), sortParameters.getPartitionID(),
- sortParameters.getSegmentId() + "", false);
+ sortParameters.getSegmentId() + "", false, false);
// Set the data file location
String[] dataFolderLocations = CarbonDataProcessorUtil.arrayAppend(storeLocations,
File.separator, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterWithBucketingImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterWithBucketingImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterWithBucketingImpl.java
index 7314b1e..851c384 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterWithBucketingImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/ParallelReadMergeSorterWithBucketingImpl.java
@@ -137,7 +137,7 @@ public class ParallelReadMergeSorterWithBucketingImpl extends AbstractMergeSorte
String[] storeLocation = CarbonDataProcessorUtil
.getLocalDataFolderLocation(sortParameters.getDatabaseName(), sortParameters.getTableName(),
String.valueOf(sortParameters.getTaskNo()), bucketId,
- sortParameters.getSegmentId() + "", false);
+ sortParameters.getSegmentId() + "", false, false);
// Set the data file location
String[] dataFolderLocation = CarbonDataProcessorUtil.arrayAppend(storeLocation, File.separator,
CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
@@ -188,7 +188,7 @@ public class ParallelReadMergeSorterWithBucketingImpl extends AbstractMergeSorte
String[] carbonDataDirectoryPath = CarbonDataProcessorUtil
.getLocalDataFolderLocation(parameters.getDatabaseName(),
parameters.getTableName(), parameters.getTaskNo(),
- parameters.getPartitionID(), parameters.getSegmentId(), false);
+ parameters.getPartitionID(), parameters.getSegmentId(), false, false);
String[] tmpLocs = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath, File.separator,
CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
parameters.setTempFileLocation(tmpLocs);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
index e140d86..ebb85b4 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeBatchParallelReadMergeSorterImpl.java
@@ -224,7 +224,7 @@ public class UnsafeBatchParallelReadMergeSorterImpl extends AbstractMergeSorter
String[] carbonDataDirectoryPath = CarbonDataProcessorUtil
.getLocalDataFolderLocation(parameters.getDatabaseName(),
parameters.getTableName(), parameters.getTaskNo(), batchCount + "",
- parameters.getSegmentId(), false);
+ parameters.getSegmentId(), false, false);
String[] tempDirs = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath,
File.separator, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
parameters.setTempFileLocation(tempDirs);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeParallelReadMergeSorterWithBucketingImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeParallelReadMergeSorterWithBucketingImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeParallelReadMergeSorterWithBucketingImpl.java
index e508654..f000619 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeParallelReadMergeSorterWithBucketingImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/sort/impl/UnsafeParallelReadMergeSorterWithBucketingImpl.java
@@ -127,7 +127,7 @@ public class UnsafeParallelReadMergeSorterWithBucketingImpl implements Sorter {
String[] storeLocation = CarbonDataProcessorUtil
.getLocalDataFolderLocation(sortParameters.getDatabaseName(), sortParameters.getTableName(),
String.valueOf(sortParameters.getTaskNo()), bucketId,
- sortParameters.getSegmentId() + "", false);
+ sortParameters.getSegmentId() + "", false, false);
// Set the data file location
String[] dataFolderLocation = CarbonDataProcessorUtil.arrayAppend(storeLocation,
File.separator, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
@@ -170,7 +170,8 @@ public class UnsafeParallelReadMergeSorterWithBucketingImpl implements Sorter {
private void setTempLocation(SortParameters parameters) {
String[] carbonDataDirectoryPath = CarbonDataProcessorUtil
.getLocalDataFolderLocation(parameters.getDatabaseName(), parameters.getTableName(),
- parameters.getTaskNo(), parameters.getPartitionID(), parameters.getSegmentId(), false);
+ parameters.getTaskNo(), parameters.getPartitionID(), parameters.getSegmentId(),
+ false, false);
String[] tmpLoc = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath, File.separator,
CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
parameters.setTempFileLocation(tmpLoc);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
index fe9257f..765e0ed 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/CarbonRowDataWriterProcessorStepImpl.java
@@ -91,7 +91,7 @@ public class CarbonRowDataWriterProcessorStepImpl extends AbstractDataLoadProces
String[] storeLocation = CarbonDataProcessorUtil
.getLocalDataFolderLocation(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), String.valueOf(configuration.getTaskNo()), partitionId,
- configuration.getSegmentId() + "", false);
+ configuration.getSegmentId() + "", false, false);
CarbonDataProcessorUtil.createLocations(storeLocation);
return storeLocation;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterBatchProcessorStepImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterBatchProcessorStepImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterBatchProcessorStepImpl.java
index c5f2479..fc4d4d2 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterBatchProcessorStepImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterBatchProcessorStepImpl.java
@@ -62,7 +62,7 @@ public class DataWriterBatchProcessorStepImpl extends AbstractDataLoadProcessorS
String[] storeLocation = CarbonDataProcessorUtil
.getLocalDataFolderLocation(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), String.valueOf(configuration.getTaskNo()), partitionId,
- configuration.getSegmentId() + "", false);
+ configuration.getSegmentId() + "", false, false);
CarbonDataProcessorUtil.createLocations(storeLocation);
return storeLocation;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterProcessorStepImpl.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterProcessorStepImpl.java b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterProcessorStepImpl.java
index aad874b..c7af420 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterProcessorStepImpl.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/newflow/steps/DataWriterProcessorStepImpl.java
@@ -69,7 +69,7 @@ public class DataWriterProcessorStepImpl extends AbstractDataLoadProcessorStep {
String[] storeLocation = CarbonDataProcessorUtil
.getLocalDataFolderLocation(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), String.valueOf(configuration.getTaskNo()), partitionId,
- configuration.getSegmentId() + "", false);
+ configuration.getSegmentId() + "", false, false);
CarbonDataProcessorUtil.createLocations(storeLocation);
return storeLocation;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/sortandgroupby/sortdata/SortParameters.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/sortandgroupby/sortdata/SortParameters.java b/processing/src/main/java/org/apache/carbondata/processing/sortandgroupby/sortdata/SortParameters.java
index 16cab07..fb2977e 100644
--- a/processing/src/main/java/org/apache/carbondata/processing/sortandgroupby/sortdata/SortParameters.java
+++ b/processing/src/main/java/org/apache/carbondata/processing/sortandgroupby/sortdata/SortParameters.java
@@ -412,7 +412,7 @@ public class SortParameters implements Serializable {
String[] carbonDataDirectoryPath = CarbonDataProcessorUtil
.getLocalDataFolderLocation(tableIdentifier.getDatabaseName(),
tableIdentifier.getTableName(), configuration.getTaskNo(),
- configuration.getPartitionId(), configuration.getSegmentId(), false);
+ configuration.getPartitionId(), configuration.getSegmentId(), false, false);
String[] sortTempDirs = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath,
File.separator, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
@@ -534,7 +534,7 @@ public class SortParameters implements Serializable {
String[] carbonDataDirectoryPath = CarbonDataProcessorUtil
.getLocalDataFolderLocation(databaseName, tableName, taskNo, partitionID, segmentId,
- isCompactionFlow);
+ isCompactionFlow, false);
String[] sortTempDirs = CarbonDataProcessorUtil.arrayAppend(carbonDataDirectoryPath,
File.separator, CarbonCommonConstants.SORT_TEMP_FILE_LOCATION);
parameters.setTempFileLocation(sortTempDirs);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/spliter/AbstractCarbonQueryExecutor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/spliter/AbstractCarbonQueryExecutor.java b/processing/src/main/java/org/apache/carbondata/processing/spliter/AbstractCarbonQueryExecutor.java
new file mode 100644
index 0000000..39d1234
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/spliter/AbstractCarbonQueryExecutor.java
@@ -0,0 +1,133 @@
+/*
+ * 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.processing.spliter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.carbondata.common.CarbonIterator;
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.dictionary.Dictionary;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.block.TableBlockInfo;
+import org.apache.carbondata.core.datastore.block.TaskBlockInfo;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
+import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
+import org.apache.carbondata.core.scan.executor.QueryExecutor;
+import org.apache.carbondata.core.scan.executor.QueryExecutorFactory;
+import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
+import org.apache.carbondata.core.scan.model.QueryDimension;
+import org.apache.carbondata.core.scan.model.QueryMeasure;
+import org.apache.carbondata.core.scan.model.QueryModel;
+import org.apache.carbondata.core.scan.result.BatchResult;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+public abstract class AbstractCarbonQueryExecutor {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(AbstractCarbonQueryExecutor.class.getName());
+ protected CarbonTable carbonTable;
+ protected QueryModel queryModel;
+ protected QueryExecutor queryExecutor;
+ protected Map<String, TaskBlockInfo> segmentMapping;
+
+ /**
+ * get executor and execute the query model.
+ *
+ * @param blockList
+ * @return
+ */
+ protected CarbonIterator<BatchResult> executeBlockList(List<TableBlockInfo> blockList)
+ throws QueryExecutionException, IOException {
+ queryModel.setTableBlockInfos(blockList);
+ this.queryExecutor = QueryExecutorFactory.getQueryExecutor(queryModel);
+ return queryExecutor.execute(queryModel);
+ }
+
+ /**
+ * Preparing of the query model.
+ *
+ * @param blockList
+ * @return
+ */
+ protected QueryModel prepareQueryModel(List<TableBlockInfo> blockList) {
+ QueryModel model = new QueryModel();
+ model.setTableBlockInfos(blockList);
+ model.setForcedDetailRawQuery(true);
+ model.setFilterExpressionResolverTree(null);
+
+ List<QueryDimension> dims = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+
+ List<CarbonDimension> dimensions =
+ carbonTable.getDimensionByTableName(carbonTable.getFactTableName());
+ for (CarbonDimension dim : dimensions) {
+ // check if dimension is deleted
+ QueryDimension queryDimension = new QueryDimension(dim.getColName());
+ queryDimension.setDimension(dim);
+ dims.add(queryDimension);
+ }
+ model.setQueryDimension(dims);
+
+ List<QueryMeasure> msrs = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+ List<CarbonMeasure> measures =
+ carbonTable.getMeasureByTableName(carbonTable.getFactTableName());
+ for (CarbonMeasure carbonMeasure : measures) {
+ // check if measure is deleted
+ QueryMeasure queryMeasure = new QueryMeasure(carbonMeasure.getColName());
+ queryMeasure.setMeasure(carbonMeasure);
+ msrs.add(queryMeasure);
+ }
+ model.setQueryMeasures(msrs);
+ model.setQueryId(System.nanoTime() + "");
+ model.setAbsoluteTableIdentifier(carbonTable.getAbsoluteTableIdentifier());
+ model.setTable(carbonTable);
+ return model;
+ }
+
+ /**
+ * Below method will be used
+ * for cleanup
+ */
+ public void finish() {
+ try {
+ queryExecutor.finish();
+ } catch (QueryExecutionException e) {
+ LOGGER.error(e, "Problem while finish: ");
+ }
+ clearDictionaryFromQueryModel();
+ }
+
+ /**
+ * This method will clear the dictionary access count after its usage is complete so
+ * that column can be deleted form LRU cache whenever memory reaches threshold
+ */
+ private void clearDictionaryFromQueryModel() {
+ if (null != queryModel) {
+ Map<String, Dictionary> columnToDictionaryMapping = queryModel.getColumnToDictionaryMapping();
+ if (null != columnToDictionaryMapping) {
+ for (Map.Entry<String, Dictionary> entry : columnToDictionaryMapping.entrySet()) {
+ CarbonUtil.clearDictionaryCache(entry.getValue());
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/spliter/CarbonSplitExecutor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/spliter/CarbonSplitExecutor.java b/processing/src/main/java/org/apache/carbondata/processing/spliter/CarbonSplitExecutor.java
new file mode 100644
index 0000000..7b724ee
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/spliter/CarbonSplitExecutor.java
@@ -0,0 +1,64 @@
+/*
+ * 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.processing.spliter;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+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.core.datastore.block.TableBlockInfo;
+import org.apache.carbondata.core.datastore.block.TaskBlockInfo;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException;
+import org.apache.carbondata.core.scan.result.iterator.PartitionSpliterRawResultIterator;
+
+/**
+ * Used to read carbon blocks when add/split partition
+ */
+public class CarbonSplitExecutor extends AbstractCarbonQueryExecutor {
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(CarbonSplitExecutor.class.getName());
+
+ public CarbonSplitExecutor(Map<String, TaskBlockInfo> segmentMapping, CarbonTable carbonTable) {
+ this.segmentMapping = segmentMapping;
+ this.carbonTable = carbonTable;
+ }
+
+ public List<PartitionSpliterRawResultIterator> processDataBlocks(String segmentId)
+ throws QueryExecutionException, IOException {
+ List<TableBlockInfo> list = null;
+ queryModel = prepareQueryModel(list);
+ List<PartitionSpliterRawResultIterator> resultList
+ = new ArrayList<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
+ TaskBlockInfo taskBlockInfo = segmentMapping.get(segmentId);
+ Set<String> taskBlockListMapping = taskBlockInfo.getTaskSet();
+ for (String task : taskBlockListMapping) {
+ list = taskBlockInfo.getTableBlockInfoList(task);
+ LOGGER.info("for task -" + task + "-block size is -" + list.size());
+ queryModel.setTableBlockInfos(list);
+ resultList.add(new PartitionSpliterRawResultIterator(executeBlockList(list)));
+ }
+ return resultList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/spliter/RowResultSpliterProcessor.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/spliter/RowResultSpliterProcessor.java b/processing/src/main/java/org/apache/carbondata/processing/spliter/RowResultSpliterProcessor.java
new file mode 100644
index 0000000..ea38a53
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/spliter/RowResultSpliterProcessor.java
@@ -0,0 +1,105 @@
+/*
+ * 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.processing.spliter;
+
+import java.util.List;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.datastore.block.SegmentProperties;
+import org.apache.carbondata.core.datastore.exception.CarbonDataWriterException;
+import org.apache.carbondata.core.datastore.row.CarbonRow;
+import org.apache.carbondata.core.datastore.row.WriteStepRowUtil;
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
+import org.apache.carbondata.processing.model.CarbonLoadModel;
+import org.apache.carbondata.processing.spliter.exception.SliceSpliterException;
+import org.apache.carbondata.processing.store.CarbonDataFileAttributes;
+import org.apache.carbondata.processing.store.CarbonFactDataHandlerColumnar;
+import org.apache.carbondata.processing.store.CarbonFactDataHandlerModel;
+import org.apache.carbondata.processing.store.CarbonFactHandler;
+import org.apache.carbondata.processing.util.CarbonDataProcessorUtil;
+
+public class RowResultSpliterProcessor {
+
+ private CarbonFactHandler dataHandler;
+ private SegmentProperties segmentProperties;
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(RowResultSpliterProcessor.class.getName());
+
+
+ public RowResultSpliterProcessor(CarbonTable carbonTable, CarbonLoadModel loadModel,
+ SegmentProperties segProp, String[] tempStoreLocation, Integer bucketId) {
+ CarbonDataProcessorUtil.createLocations(tempStoreLocation);
+ this.segmentProperties = segProp;
+ String tableName = carbonTable.getFactTableName();
+ CarbonFactDataHandlerModel carbonFactDataHandlerModel =
+ CarbonFactDataHandlerModel.getCarbonFactDataHandlerModel(loadModel, carbonTable,
+ segProp, tableName, tempStoreLocation);
+ CarbonDataFileAttributes carbonDataFileAttributes =
+ new CarbonDataFileAttributes(Integer.parseInt(loadModel.getTaskNo()),
+ loadModel.getFactTimeStamp());
+ carbonFactDataHandlerModel.setCarbonDataFileAttributes(carbonDataFileAttributes);
+ carbonFactDataHandlerModel.setBucketId(bucketId);
+ //Note: set compaction flow just to convert decimal type
+ carbonFactDataHandlerModel.setCompactionFlow(true);
+ dataHandler = new CarbonFactDataHandlerColumnar(carbonFactDataHandlerModel);
+ }
+
+ public boolean execute(List<Object[]> resultList) {
+ boolean splitStatus;
+ boolean isDataPresent = false;
+
+ try {
+ if (!isDataPresent) {
+ dataHandler.initialise();
+ isDataPresent = true;
+ }
+ for (Object[] row: resultList) {
+ addRow(row);
+ }
+ if (isDataPresent)
+ {
+ this.dataHandler.finish();
+ }
+ splitStatus = true;
+ } catch (SliceSpliterException e) {
+ LOGGER.error(e, e.getMessage());
+ LOGGER.error("Exception in split partition" + e.getMessage());
+ splitStatus = false;
+ } finally {
+ try {
+ if (isDataPresent) {
+ this.dataHandler.closeHandler();
+ }
+ } catch (Exception e) {
+ LOGGER.error("Exception while closing the handler in partition spliter" + e.getMessage());
+ splitStatus = false;
+ }
+ }
+ return splitStatus;
+ }
+
+ private void addRow(Object[] carbonTuple) throws SliceSpliterException {
+ CarbonRow row = WriteStepRowUtil.fromMergerRow(carbonTuple, segmentProperties);
+ try {
+ this.dataHandler.addDataToStore(row);
+ } catch (CarbonDataWriterException e) {
+ throw new SliceSpliterException("Problem in writing rows when add/split the partition", e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/processing/src/main/java/org/apache/carbondata/processing/spliter/exception/SliceSpliterException.java
----------------------------------------------------------------------
diff --git a/processing/src/main/java/org/apache/carbondata/processing/spliter/exception/SliceSpliterException.java b/processing/src/main/java/org/apache/carbondata/processing/spliter/exception/SliceSpliterException.java
new file mode 100644
index 0000000..17e679a
--- /dev/null
+++ b/processing/src/main/java/org/apache/carbondata/processing/spliter/exception/SliceSpliterException.java
@@ -0,0 +1,78 @@
+/*
+ * 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.processing.spliter.exception;
+
+import java.util.Locale;
+
+public class SliceSpliterException extends Exception {
+
+ /**
+ * default serial version ID.
+ */
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * The Error message.
+ */
+ private String msg = "";
+
+ /**
+ * Constructor
+ *
+ * @param msg The error message for this exception.
+ */
+ public SliceSpliterException(String msg) {
+ super(msg);
+ this.msg = msg;
+ }
+
+ /**
+ * Constructor
+ *
+ * @param msg The error message for this exception.
+ */
+ public SliceSpliterException(String msg, Throwable t) {
+ super(msg, t);
+ this.msg = msg;
+ }
+
+ /**
+ * 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 public String getLocalizedMessage() {
+ return super.getLocalizedMessage();
+ }
+
+ /**
+ * getMessage
+ */
+ public String getMessage() {
+ return this.msg;
+ }
+}
[3/3] carbondata git commit: [CARBONDATA-940] alter table add/split
partition for spark 2.1
Posted by ja...@apache.org.
[CARBONDATA-940] alter table add/split partition for spark 2.1
add/split partition function
This closes #1192
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/874764f9
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/874764f9
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/874764f9
Branch: refs/heads/master
Commit: 874764f95629c98d8f98a424ca742614e38f704f
Parents: 3efd330
Author: lionelcao <wh...@gmail.com>
Authored: Wed Jul 19 14:36:18 2017 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Thu Aug 17 09:28:54 2017 +0800
----------------------------------------------------------------------
conf/carbon.properties.template | 3 +
.../core/constants/CarbonCommonConstants.java | 9 +
.../apache/carbondata/core/locks/LockUsage.java | 1 +
.../core/metadata/schema/PartitionInfo.java | 25 +
.../filter/partition/PartitionFilterUtil.java | 2 +-
.../PartitionSpliterRawResultIterator.java | 72 +++
.../core/scan/wrappers/ByteArrayWrapper.java | 4 +-
.../core/util/path/CarbonTablePath.java | 18 +
.../src/main/resources/partition_data.csv | 25 -
.../examples/CarbonPartitionExample.scala | 75 ++-
.../hadoop/api/CarbonTableInputFormat.java | 150 +++++-
.../hadoop/util/CarbonInputFormatUtil.java | 8 +
.../src/test/resources/partition_data.csv | 28 +
.../TestDataLoadingForPartitionTable.scala | 4 +-
.../carbondata/spark/load/CarbonLoaderUtil.java | 4 +-
.../spark/partition/SplitPartitionCallable.java | 41 ++
.../org/apache/carbondata/spark/KeyVal.scala | 8 +
.../load/DataLoadProcessorStepOnSpark.scala | 2 +-
.../spark/rdd/AlterTableSplitPartitionRDD.scala | 146 ++++++
.../carbondata/spark/rdd/CarbonMergerRDD.scala | 8 +-
.../spark/rdd/CarbonScanPartitionRDD.scala | 269 ++++++++++
.../spark/rdd/DataManagementFunc.scala | 47 +-
.../spark/rdd/NewCarbonDataLoadRDD.scala | 6 +-
.../spark/rdd/PartitionSplitter.scala | 104 ++++
.../carbondata/spark/util/CommonUtil.scala | 40 +-
.../execution/command/carbonTableSchema.scala | 16 +
.../org/apache/spark/util/PartitionUtils.scala | 135 +++++
.../spark/rdd/CarbonDataRDDFactory.scala | 99 +++-
.../execution/command/carbonTableSchema.scala | 147 +++++-
.../sql/parser/CarbonSpark2SqlParser.scala | 26 +-
.../spark/sql/parser/CarbonSparkSqlParser.scala | 2 +-
.../partition/TestAlterPartitionTable.scala | 523 +++++++++++++++++++
.../merger/CompactionResultSortProcessor.java | 3 +-
.../newflow/DataLoadProcessBuilder.java | 2 +-
.../sort/impl/ParallelReadMergeSorterImpl.java | 2 +-
...arallelReadMergeSorterWithBucketingImpl.java | 4 +-
.../UnsafeBatchParallelReadMergeSorterImpl.java | 2 +-
...arallelReadMergeSorterWithBucketingImpl.java | 5 +-
.../CarbonRowDataWriterProcessorStepImpl.java | 2 +-
.../steps/DataWriterBatchProcessorStepImpl.java | 2 +-
.../steps/DataWriterProcessorStepImpl.java | 2 +-
.../sortandgroupby/sortdata/SortParameters.java | 4 +-
.../spliter/AbstractCarbonQueryExecutor.java | 133 +++++
.../processing/spliter/CarbonSplitExecutor.java | 64 +++
.../spliter/RowResultSpliterProcessor.java | 105 ++++
.../exception/SliceSpliterException.java | 78 +++
.../store/CarbonFactDataHandlerModel.java | 2 +
.../util/CarbonDataProcessorUtil.java | 12 +-
48 files changed, 2360 insertions(+), 109 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/conf/carbon.properties.template
----------------------------------------------------------------------
diff --git a/conf/carbon.properties.template b/conf/carbon.properties.template
index b5f5101..de8a4ee 100644
--- a/conf/carbon.properties.template
+++ b/conf/carbon.properties.template
@@ -42,6 +42,9 @@ carbon.enableXXHash=true
#carbon.max.level.cache.size=-1
#enable prefetch of data during merge sort while reading data from sort temp files in data loading
#carbon.merge.sort.prefetch=true
+######## Alter Partition Configuration ########
+#Number of cores to be used while alter partition
+carbon.number.of.cores.while.alterPartition=2
######## Compaction Configuration ########
#Number of cores to be used while compacting
carbon.number.of.cores.while.compacting=2
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index 14258f8..74a08ec 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -306,6 +306,10 @@ public final class CarbonCommonConstants {
@CarbonProperty
public static final String NUM_CORES_COMPACTING = "carbon.number.of.cores.while.compacting";
/**
+ * Number of cores to be used while alter partition
+ */
+ public static final String NUM_CORES_ALT_PARTITION = "carbon.number.of.cores.while.altPartition";
+ /**
* Number of cores to be used for block sort
*/
@CarbonProperty
@@ -967,6 +971,11 @@ public final class CarbonCommonConstants {
public static final String COMPACTION_KEY_WORD = "COMPACTION";
/**
+ * Indicates alter partition
+ */
+ public static String ALTER_PARTITION_KEY_WORD = "ALTER_PARTITION";
+
+ /**
* hdfs temporary directory key
*/
@CarbonProperty
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/locks/LockUsage.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/locks/LockUsage.java b/core/src/main/java/org/apache/carbondata/core/locks/LockUsage.java
index 046fff5..1738c64 100644
--- a/core/src/main/java/org/apache/carbondata/core/locks/LockUsage.java
+++ b/core/src/main/java/org/apache/carbondata/core/locks/LockUsage.java
@@ -26,6 +26,7 @@ public class LockUsage {
public static final String METADATA_LOCK = "meta.lock";
public static final String COMPACTION_LOCK = "compaction.lock";
public static final String SYSTEMLEVEL_COMPACTION_LOCK = "system_level_compaction.lock";
+ public static final String ALTER_PARTITION_LOCK = "alter_partition.lock";
public static final String TABLE_STATUS_LOCK = "tablestatus.lock";
public static final String TABLE_UPDATE_STATUS_LOCK = "tableupdatestatus.lock";
public static final String DELETE_SEGMENT_LOCK = "delete_segment.lock";
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/metadata/schema/PartitionInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/PartitionInfo.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/PartitionInfo.java
index 39d7f3f..4b0bc3e 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/PartitionInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/PartitionInfo.java
@@ -67,6 +67,31 @@ public class PartitionInfo implements Serializable {
this.partitionIds = new ArrayList<>();
}
+ /**
+ * add partition means split default partition, add in last directly
+ */
+ public void addPartition(int addPartitionCount) {
+ for (int i = 0; i < addPartitionCount; i++) {
+ partitionIds.add(++MAX_PARTITION);
+ numPartitions++;
+ }
+ }
+
+ /**
+ * e.g. original partition[0,1,2,3,4,5]
+ * split partition 2 to partition 6,7,8 (will not reuse 2)
+ * then sourcePartitionId is 2, newPartitionNumbers is 3
+ * @param sourcePartitionIndex
+ * @param newPartitionNumbers
+ */
+ public void splitPartition(int sourcePartitionIndex, int newPartitionNumbers) {
+ partitionIds.remove(sourcePartitionIndex);
+ for (int i = 0; i < newPartitionNumbers; i++) {
+ partitionIds.add(sourcePartitionIndex + i, ++MAX_PARTITION);
+ }
+ numPartitions = numPartitions - 1 + newPartitionNumbers;
+ }
+
public List<ColumnSchema> getColumnSchemaList() {
return columnSchemaList;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/scan/filter/partition/PartitionFilterUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/filter/partition/PartitionFilterUtil.java b/core/src/main/java/org/apache/carbondata/core/scan/filter/partition/PartitionFilterUtil.java
index d040c1b..9671199 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/filter/partition/PartitionFilterUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/filter/partition/PartitionFilterUtil.java
@@ -107,7 +107,7 @@ public class PartitionFilterUtil {
}
}
} else {
- // LessThanEqualTo(<)
+ // LessThan(<)
outer4:
for (int i = 0; i < partitions; i++) {
for (String value : listInfo.get(i)) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/scan/result/iterator/PartitionSpliterRawResultIterator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/iterator/PartitionSpliterRawResultIterator.java b/core/src/main/java/org/apache/carbondata/core/scan/result/iterator/PartitionSpliterRawResultIterator.java
new file mode 100644
index 0000000..553f85e
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/scan/result/iterator/PartitionSpliterRawResultIterator.java
@@ -0,0 +1,72 @@
+/*
+ * 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.core.scan.result.iterator;
+
+import org.apache.carbondata.common.CarbonIterator;
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.scan.result.BatchResult;
+
+public class PartitionSpliterRawResultIterator extends CarbonIterator<Object[]> {
+
+ private CarbonIterator<BatchResult> iterator;
+ private BatchResult batch;
+ private int counter;
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(PartitionSpliterRawResultIterator.class.getName());
+
+ public PartitionSpliterRawResultIterator(CarbonIterator<BatchResult> iterator) {
+ this.iterator = iterator;
+ }
+
+ @Override public boolean hasNext() {
+ if (null == batch || checkBatchEnd(batch)) {
+ if (iterator.hasNext()) {
+ batch = iterator.next();
+ counter = 0;
+ } else {
+ return false;
+ }
+ }
+
+ return !checkBatchEnd(batch);
+ }
+
+ @Override public Object[] next() {
+ if (batch == null) {
+ batch = iterator.next();
+ }
+ if (!checkBatchEnd(batch)) {
+ return batch.getRawRow(counter++);
+ } else {
+ batch = iterator.next();
+ counter = 0;
+ }
+ return batch.getRawRow(counter++);
+ }
+
+ /**
+ * To check if the batch is processed completely
+ * @param batch
+ * @return
+ */
+ private boolean checkBatchEnd(BatchResult batch) {
+ return !(counter < batch.getSize());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/scan/wrappers/ByteArrayWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/wrappers/ByteArrayWrapper.java b/core/src/main/java/org/apache/carbondata/core/scan/wrappers/ByteArrayWrapper.java
index 93bf8eb..2f981b5 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/wrappers/ByteArrayWrapper.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/wrappers/ByteArrayWrapper.java
@@ -16,13 +16,15 @@
*/
package org.apache.carbondata.core.scan.wrappers;
+import java.io.Serializable;
+
import org.apache.carbondata.core.util.ByteUtil.UnsafeComparer;
/**
* This class will store the dimension column data when query is executed
* This can be used as a key for aggregation
*/
-public class ByteArrayWrapper implements Comparable<ByteArrayWrapper> {
+public class ByteArrayWrapper implements Comparable<ByteArrayWrapper>, Serializable {
/**
* to store key which is generated using
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
index 7b8904c..fb9b783 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/path/CarbonTablePath.java
@@ -303,6 +303,20 @@ public class CarbonTablePath extends Path {
}
}
+ public String getCarbonIndexFilePath(String taskId, String partitionId, String segmentId,
+ int batchNo, String bucketNumber, String timeStamp,
+ ColumnarFormatVersion columnarFormatVersion) {
+ switch (columnarFormatVersion) {
+ case V1:
+ case V2:
+ return getCarbonIndexFilePath(taskId, partitionId, segmentId, bucketNumber);
+ default:
+ String segmentDir = getSegmentDir(partitionId, segmentId);
+ return segmentDir + File.separator + getCarbonIndexFileName(Integer.parseInt(taskId),
+ Integer.parseInt(bucketNumber), batchNo, timeStamp);
+ }
+ }
+
private static String getCarbonIndexFileName(String taskNo, int bucketNumber,
String factUpdatedtimeStamp) {
return taskNo + "-" + bucketNumber + "-" + factUpdatedtimeStamp + INDEX_FILE_EXT;
@@ -537,6 +551,10 @@ public class CarbonTablePath extends Path {
return Integer.parseInt(taskNo.split(BATCH_PREFIX)[0]);
}
+ public static int getBatchNoFromTaskNo(String taskNo) {
+ return Integer.parseInt(taskNo.split(BATCH_PREFIX)[1]);
+ }
+
/**
* Gets the file name from file path
*/
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/examples/spark2/src/main/resources/partition_data.csv
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/resources/partition_data.csv b/examples/spark2/src/main/resources/partition_data.csv
deleted file mode 100644
index 2d521b7..0000000
--- a/examples/spark2/src/main/resources/partition_data.csv
+++ /dev/null
@@ -1,25 +0,0 @@
-vin,logdate,phonenumber,country,area
-A42158424831,2016/02/12,125371341,China,Asia
-A42158473831,2016/01/12,125371342,China,Asia
-A42152474832,2016/02/12,125371343,US,America
-A42151477823,2016/12/12,125371344,China,Asia
-A42158474135,2016/02/15,125371345,Japan,Asia
-A42258434831,2016/12/12,125371346,China,Asia
-A42158475831,2016/05/12,125371347,UK,Europe
-A41158494830,2015/07/12,225371348,China,Asia
-A42158424831,2015/02/12,225371349,China,Asia
-A42158473830,2014/01/12,225371310,China,Asia
-A42152474830,2013/02/12,325371311,US,America
-A42151477823,2012/12/12,425371312,China,Asia
-A42158474133,2012/02/15,325371313,Japan,Asia
-A42258434835,2013/12/12,525371314,China,Asia
-A42158475836,2014/05/12,625371315,UK,Europe
-A41158494838,2015/07/12,525371316,China,Asia
-A42158424833,2016/02/12,425371317,China,Asia
-A42158473832,2017/01/12,325371318,China,Asia
-A42152474834,2011/02/12,225371319,US,America
-A42151477824,2012/12/12,225371320,China,Asia
-A42158474137,2013/02/15,325371321,Japan,Asia
-A42258434837,2014/12/12,25371322,China,Asia
-A42158475838,2014/05/12,425371323,UK,Europe
-A41158494839,2016/07/12,625371324,China,Asia
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
index 8a01ba1..ca0501c 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
@@ -35,7 +35,7 @@ object CarbonPartitionExample {
val storeLocation = s"$rootPath/examples/spark2/target/store"
val warehouse = s"$rootPath/examples/spark2/target/warehouse"
val metastoredb = s"$rootPath/examples/spark2/target"
- val testData = s"$rootPath/examples/spark2/src/main/resources/partition_data.csv"
+ val testData = s"$rootPath/integration/spark-common-test/src/test/resources/partition_data.csv"
CarbonProperties.getInstance()
.addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
@@ -51,34 +51,57 @@ object CarbonPartitionExample {
spark.sparkContext.setLogLevel("WARN")
- // none partition table
+ // range partition with bucket defined
spark.sql("DROP TABLE IF EXISTS t0")
spark.sql("""
| CREATE TABLE IF NOT EXISTS t0
| (
+ | id Int,
| vin String,
- | logdate Timestamp,
| phonenumber Long,
| country String,
- | area String
+ | area String,
+ | salary Int
| )
+ | PARTITIONED BY (logdate Timestamp)
| STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
+ | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01',
+ | 'BUCKETNUMBER'='3',
+ | 'BUCKETCOLUMNS'='vin')
""".stripMargin)
- // range partition
+ // none partition table
spark.sql("DROP TABLE IF EXISTS t1")
spark.sql("""
| CREATE TABLE IF NOT EXISTS t1
| (
+ | id Int,
| vin String,
+ | logdate Timestamp,
| phonenumber Long,
| country String,
| area String
| )
- | PARTITIONED BY (logdate Timestamp)
| STORED BY 'carbondata'
- | TBLPROPERTIES('PARTITION_TYPE'='RANGE',
- | 'RANGE_INFO'='2014/01/01, 2015/01/01, 2016/01/01')
+ """.stripMargin)
+
+ // list partition
+ spark.sql("DROP TABLE IF EXISTS t2")
+ spark.sql("""
+ | CREATE TABLE IF NOT EXISTS t2
+ | (
+ | id Int,
+ | vin String,
+ | logdate Timestamp,
+ | phonenumber Long,
+ | country String,
+ | salary Int
+ | )
+ | PARTITIONED BY (area String)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES('PARTITION_TYPE'='LIST',
+ | 'LIST_INFO'='Asia, America, Europe', 'DICTIONARY_EXCLUDE' ='area')
""".stripMargin)
// hash partition
@@ -86,10 +109,12 @@ object CarbonPartitionExample {
spark.sql("""
| CREATE TABLE IF NOT EXISTS t3
| (
+ | id Int,
| logdate Timestamp,
| phonenumber Long,
| country String,
- | area String
+ | area String,
+ | salary Int
| )
| PARTITIONED BY (vin String)
| STORED BY 'carbondata'
@@ -101,17 +126,40 @@ object CarbonPartitionExample {
spark.sql("""
| CREATE TABLE IF NOT EXISTS t5
| (
+ | id Int,
| vin String,
| logdate Timestamp,
| phonenumber Long,
- | area String
+ | area String,
+ | salary Int
|)
| PARTITIONED BY (country String)
| STORED BY 'carbondata'
| TBLPROPERTIES('PARTITION_TYPE'='LIST',
- | 'LIST_INFO'='(China,United States),UK ,japan,(Canada,Russia), South Korea ')
+ | 'LIST_INFO'='(China, US),UK ,Japan,(Canada,Russia, Good, NotGood), Korea ')
""".stripMargin)
+ // load data into partition table
+ spark.sql(s"""
+ LOAD DATA LOCAL INPATH '$testData' into table t0 options('BAD_RECORDS_ACTION'='FORCE')
+ """)
+ spark.sql(s"""
+ LOAD DATA LOCAL INPATH '$testData' into table t5 options('BAD_RECORDS_ACTION'='FORCE')
+ """)
+
+ // alter list partition table t5 to add a partition
+ spark.sql(s"""Alter table t5 add partition ('OutSpace')""".stripMargin)
+ // alter list partition table t5 to split partition 4 into 3 independent partition
+ spark.sql(
+ s"""
+ Alter table t5 split partition(4) into ('Canada', 'Russia', '(Good, NotGood)')
+ """.stripMargin)
+
+ spark.sql("""select * from t5 where country = 'Good' """).show(100, false)
+
+ spark.sql("select * from t0 order by salary ").show(100, false)
+ spark.sql("select * from t5 order by salary ").show(100, false)
+
// hive partition table
spark.sql("DROP TABLE IF EXISTS t7")
spark.sql("""
@@ -130,6 +178,7 @@ object CarbonPartitionExample {
spark.sql(s"CREATE DATABASE partitionDB")
spark.sql(s"""
| CREATE TABLE IF NOT EXISTS partitionDB.t9(
+ | id Int,
| logdate Timestamp,
| phonenumber Int,
| country String,
@@ -145,11 +194,11 @@ object CarbonPartitionExample {
// show partitions
try {
- spark.sql("""SHOW PARTITIONS t0""").show(100, false)
+ spark.sql("""SHOW PARTITIONS t1""").show(100, false)
} catch {
case ex: AnalysisException => LOGGER.error(ex.getMessage())
}
- spark.sql("""SHOW PARTITIONS t1""").show(100, false)
+ spark.sql("""SHOW PARTITIONS t0""").show(100, false)
spark.sql("""SHOW PARTITIONS t3""").show(100, false)
spark.sql("""SHOW PARTITIONS t5""").show(100, false)
spark.sql("""SHOW PARTITIONS t7""").show(100, false)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
index 19e264b..dc4e76a 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonTableInputFormat.java
@@ -51,8 +51,6 @@ import org.apache.carbondata.core.scan.filter.FilterExpressionProcessor;
import org.apache.carbondata.core.scan.filter.resolver.FilterResolverIntf;
import org.apache.carbondata.core.scan.model.CarbonQueryPlan;
import org.apache.carbondata.core.scan.model.QueryModel;
-import org.apache.carbondata.core.scan.partition.PartitionUtil;
-import org.apache.carbondata.core.scan.partition.Partitioner;
import org.apache.carbondata.core.stats.QueryStatistic;
import org.apache.carbondata.core.stats.QueryStatisticsConstants;
import org.apache.carbondata.core.stats.QueryStatisticsRecorder;
@@ -64,6 +62,7 @@ import org.apache.carbondata.core.util.DataTypeConverter;
import org.apache.carbondata.core.util.DataTypeConverterImpl;
import org.apache.carbondata.core.util.path.CarbonStorePath;
import org.apache.carbondata.core.util.path.CarbonTablePath;
+import org.apache.carbondata.hadoop.CarbonInputSplit;
import org.apache.carbondata.hadoop.CarbonMultiBlockSplit;
import org.apache.carbondata.hadoop.CarbonProjection;
import org.apache.carbondata.hadoop.CarbonRecordReader;
@@ -101,6 +100,7 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
"mapreduce.input.carboninputformat.segmentnumbers";
// comma separated list of input files
public static final String INPUT_FILES = "mapreduce.input.carboninputformat.files";
+ public static final String ALTER_PARTITION_ID = "mapreduce.input.carboninputformat.partitionid";
private static final Log LOG = LogFactory.getLog(CarbonTableInputFormat.class);
private static final String FILTER_PREDICATE =
"mapreduce.input.carboninputformat.filter.predicate";
@@ -164,6 +164,11 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
throws IOException {
configuration.set(FileInputFormat.INPUT_DIR, tablePath);
}
+
+ public static void setPartitionIdList(Configuration configuration, List<String> partitionIds) {
+ configuration.set(ALTER_PARTITION_ID, partitionIds.toString());
+ }
+
/**
* It sets unresolved filter expression.
*
@@ -285,23 +290,17 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
if (null == carbonTable) {
throw new IOException("Missing/Corrupt schema file for table.");
}
-
+ PartitionInfo partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName());
CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
// prune partitions for filter query on partition table
BitSet matchedPartitions = null;
- if (null != filter) {
- PartitionInfo partitionInfo = carbonTable.getPartitionInfo(carbonTable.getFactTableName());
- if (null != partitionInfo) {
- Partitioner partitioner = PartitionUtil.getPartitioner(partitionInfo);
- matchedPartitions = new FilterExpressionProcessor()
- .getFilteredPartitions(filter, partitionInfo);
+ if (partitionInfo != null) {
+ matchedPartitions = setMatchedPartitions(null, filter, partitionInfo);
+ if (matchedPartitions != null) {
if (matchedPartitions.cardinality() == 0) {
- // no partition is required
return new ArrayList<InputSplit>();
- }
- if (matchedPartitions.cardinality() == partitioner.numPartitions()) {
- // all partitions are required, no need to prune partitions
+ } else if (matchedPartitions.cardinality() == partitionInfo.getNumPartitions()) {
matchedPartitions = null;
}
}
@@ -310,7 +309,8 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
FilterResolverIntf filterInterface = CarbonInputFormatUtil.resolveFilter(filter, identifier);
// do block filtering and get split
- List<InputSplit> splits = getSplits(job, filterInterface, validSegments, matchedPartitions);
+ List<InputSplit> splits =
+ getSplits(job, filterInterface, validSegments, matchedPartitions, partitionInfo, null);
// pass the invalid segment to task side in order to remove index entry in task side
if (invalidSegments.size() > 0) {
for (InputSplit split : splits) {
@@ -323,6 +323,85 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
}
/**
+ * Read data in one segment. For alter table partition statement
+ * @param job
+ * @param targetSegment
+ * @param oldPartitionIdList get old partitionId before partitionInfo was changed
+ * @return
+ * @throws IOException
+ */
+ public List<InputSplit> getSplitsOfOneSegment(JobContext job, String targetSegment,
+ List<Integer> oldPartitionIdList, PartitionInfo partitionInfo)
+ throws IOException {
+ AbsoluteTableIdentifier identifier = getAbsoluteTableIdentifier(job.getConfiguration());
+ List<String> invalidSegments = new ArrayList<>();
+ List<UpdateVO> invalidTimestampsList = new ArrayList<>();
+
+ List<String> segmentList = new ArrayList<>();
+ segmentList.add(targetSegment);
+ setSegmentsToAccess(job.getConfiguration(), segmentList);
+ try {
+
+ // process and resolve the expression
+ Expression filter = getFilterPredicates(job.getConfiguration());
+ CarbonTable carbonTable = getOrCreateCarbonTable(job.getConfiguration());
+ // this will be null in case of corrupt schema file.
+ if (null == carbonTable) {
+ throw new IOException("Missing/Corrupt schema file for table.");
+ }
+
+ CarbonInputFormatUtil.processFilterExpression(filter, carbonTable);
+
+ // prune partitions for filter query on partition table
+ String partitionIds = job.getConfiguration().get(ALTER_PARTITION_ID);
+ BitSet matchedPartitions = null;
+ if (partitionInfo != null) {
+ matchedPartitions = setMatchedPartitions(partitionIds, filter, partitionInfo);
+ if (matchedPartitions != null) {
+ if (matchedPartitions.cardinality() == 0) {
+ return new ArrayList<InputSplit>();
+ } else if (matchedPartitions.cardinality() == partitionInfo.getNumPartitions()) {
+ matchedPartitions = null;
+ }
+ }
+ }
+
+ FilterResolverIntf filterInterface = CarbonInputFormatUtil.resolveFilter(filter, identifier);
+ // do block filtering and get split
+ List<InputSplit> splits = getSplits(job, filterInterface, segmentList, matchedPartitions,
+ partitionInfo, oldPartitionIdList);
+ // pass the invalid segment to task side in order to remove index entry in task side
+ if (invalidSegments.size() > 0) {
+ for (InputSplit split : splits) {
+ ((CarbonInputSplit) split).setInvalidSegments(invalidSegments);
+ ((CarbonInputSplit) split).setInvalidTimestampRange(invalidTimestampsList);
+ }
+ }
+ return splits;
+ } catch (IOException e) {
+ throw new RuntimeException("Can't get splits of the target segment ", e);
+ }
+ }
+
+ private BitSet setMatchedPartitions(String partitionIds, Expression filter,
+ PartitionInfo partitionInfo) {
+ BitSet matchedPartitions = null;
+ if (null != partitionIds) {
+ String[] partList = partitionIds.replace("[", "").replace("]", "").split(",");
+ // only one partitionId in current alter table statement
+ matchedPartitions = new BitSet(Integer.parseInt(partList[0]));
+ for (String partitionId : partList) {
+ matchedPartitions.set(Integer.parseInt(partitionId));
+ }
+ } else {
+ if (null != filter) {
+ matchedPartitions =
+ new FilterExpressionProcessor().getFilteredPartitions(filter, partitionInfo);
+ }
+ }
+ return matchedPartitions;
+ }
+ /**
* {@inheritDoc}
* Configurations FileInputFormat.INPUT_DIR, CarbonInputFormat.INPUT_SEGMENT_NUMBERS
* are used to get table path to read.
@@ -331,7 +410,8 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
* @throws IOException
*/
private List<InputSplit> getSplits(JobContext job, FilterResolverIntf filterResolver,
- List<String> validSegments, BitSet matchedPartitions) throws IOException {
+ List<String> validSegments, BitSet matchedPartitions, PartitionInfo partitionInfo,
+ List<Integer> oldPartitionIdList) throws IOException {
List<InputSplit> result = new LinkedList<InputSplit>();
UpdateVO invalidBlockVOForSegmentId = null;
@@ -344,10 +424,10 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
isIUDTable = (updateStatusManager.getUpdateStatusDetails().length != 0);
- //for each segment fetch blocks matching filter in Driver BTree
+ // for each segment fetch blocks matching filter in Driver BTree
List<org.apache.carbondata.hadoop.CarbonInputSplit> dataBlocksOfSegment =
getDataBlocksOfSegment(job, absoluteTableIdentifier, filterResolver, matchedPartitions,
- validSegments);
+ validSegments, partitionInfo, oldPartitionIdList);
for (org.apache.carbondata.hadoop.CarbonInputSplit inputSplit : dataBlocksOfSegment) {
// Get the UpdateVO for those tables on which IUD operations being performed.
@@ -395,7 +475,8 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
*/
private List<org.apache.carbondata.hadoop.CarbonInputSplit> getDataBlocksOfSegment(JobContext job,
AbsoluteTableIdentifier absoluteTableIdentifier, FilterResolverIntf resolver,
- BitSet matchedPartitions, List<String> segmentIds) throws IOException {
+ BitSet matchedPartitions, List<String> segmentIds, PartitionInfo partitionInfo,
+ List<Integer> oldPartitionIdList) throws IOException {
QueryStatisticsRecorder recorder = CarbonTimeStatisticsFactory.createDriverRecorder();
QueryStatistic statistic = new QueryStatistic();
@@ -409,17 +490,34 @@ public class CarbonTableInputFormat<T> extends FileInputFormat<Void, T> {
List<Blocklet> prunedBlocklets = blockletMap.prune(segmentIds, resolver);
List<org.apache.carbondata.hadoop.CarbonInputSplit> resultFilterredBlocks = new ArrayList<>();
+ int partitionIndex = 0;
+ List<Integer> partitionIdList = new ArrayList<>();
+ if (partitionInfo != null) {
+ partitionIdList = partitionInfo.getPartitionIds();
+ }
for (Blocklet blocklet : prunedBlocklets) {
- int taskId = CarbonTablePath.DataFileUtil.getTaskIdFromTaskNo(
+ int partitionId = CarbonTablePath.DataFileUtil.getTaskIdFromTaskNo(
CarbonTablePath.DataFileUtil.getTaskNo(blocklet.getPath().toString()));
- // matchedPartitions variable will be null in two cases as follows
- // 1. the table is not a partition table
- // 2. the table is a partition table, and all partitions are matched by query
- // for partition table, the task id of carbaondata file name is the partition id.
- // if this partition is not required, here will skip it.
- if (matchedPartitions == null || matchedPartitions.get(taskId)) {
- resultFilterredBlocks.add(convertToCarbonInputSplit(blocklet));
+ // OldPartitionIdList is only used in alter table partition command because it change
+ // partition info first and then read data.
+ // For other normal query should use newest partitionIdList
+ if (partitionInfo != null) {
+ if (oldPartitionIdList != null) {
+ partitionIndex = oldPartitionIdList.indexOf(partitionId);
+ } else {
+ partitionIndex = partitionIdList.indexOf(partitionId);
+ }
+ }
+ if (partitionIndex != -1) {
+ // matchedPartitions variable will be null in two cases as follows
+ // 1. the table is not a partition table
+ // 2. the table is a partition table, and all partitions are matched by query
+ // for partition table, the task id of carbaondata file name is the partition id.
+ // if this partition is not required, here will skip it.
+ if (matchedPartitions == null || matchedPartitions.get(partitionIndex)) {
+ resultFilterredBlocks.add(convertToCarbonInputSplit(blocklet));
+ }
}
}
statistic
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
index 8269757..0dc79fa 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/util/CarbonInputFormatUtil.java
@@ -85,6 +85,14 @@ public class CarbonInputFormatUtil {
return carbonInputFormat;
}
+ public static <V> CarbonTableInputFormat<V> createCarbonTableInputFormat(
+ AbsoluteTableIdentifier identifier, List<String> partitionId, Job job) throws IOException {
+ CarbonTableInputFormat<V> carbonTableInputFormat = new CarbonTableInputFormat<>();
+ carbonTableInputFormat.setPartitionIdList(job.getConfiguration(), partitionId);
+ FileInputFormat.addInputPath(job, new Path(identifier.getTablePath()));
+ return carbonTableInputFormat;
+ }
+
private static void addQueryMeasure(CarbonQueryPlan plan, int order, CarbonMeasure measure) {
QueryMeasure queryMeasure = new QueryMeasure(measure.getColName());
queryMeasure.setQueryOrder(order);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common-test/src/test/resources/partition_data.csv
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/resources/partition_data.csv b/integration/spark-common-test/src/test/resources/partition_data.csv
new file mode 100644
index 0000000..ec20d92
--- /dev/null
+++ b/integration/spark-common-test/src/test/resources/partition_data.csv
@@ -0,0 +1,28 @@
+id,vin,logdate,phonenumber,country,area,salary
+1,A42158424831,2016/02/12,125371341,China,Asia,10000
+2,A42158473831,2016/01/12,125371342,China,Asia,10001
+3,A42152474832,2016/02/12,125371343,US,America,10002
+4,A42151477823,2016/12/12,125371344,China,OutSpace,10003
+5,A42158474135,2016/02/15,125371345,Japan,OutSpace,10004
+6,A42258434831,2016/12/12,125371346,China,Asia,10005
+7,A42158475831,2016/05/12,125371347,UK,OutSpace,10006
+8,A41158494830,2015/07/12,225371348,China,Asia,10007
+9,A42158424831,2015/02/12,225371349,China,OutSpace,10008
+10,A42158473830,2014/01/12,225371310,China,Asia,10009
+11,A42152474830,2013/02/12,325371311,US,America,10010
+12,A42151477823,2012/12/12,425371312,China,Asia,10011
+13,A42158474133,2012/02/15,325371313,Japan,Asia,10012
+14,A42258434835,2013/12/12,525371314,China,Asia,10013
+15,A42158475836,2014/05/12,625371315,UK,OutSpace,10014
+16,A41158494838,2015/07/12,525371316,China,Asia,10015
+17,A42158424833,2016/02/12,425371317,China,Asia,10016
+18,A42158473832,2017/01/12,325371318,China,Asia,10017
+19,A42152474834,2011/02/12,225371319,US,America,10018
+20,A42151477824,2012/12/12,225371320,China,Asia,10019
+21,A42158474137,2013/02/15,325371321,Japan,Asia,10020
+22,A42258434837,2014/12/12,25371322,China,Asia,10021
+23,A42158475838,2014/05/12,425371323,UK,OutSpace,10022
+24,A41158494839,2016/07/12,625371324,China,Asia,10023
+25,A41158494840,2016/07/12,626381324,Good,OutSpace,10024
+26,A41158494843,2016/07/12,625378824,NotGood,OutSpace,10025
+27,A41158494999,2016/07/12,625378824,Other,,10026
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
index f7ac87c..39777de 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/partition/TestDataLoadingForPartitionTable.scala
@@ -59,11 +59,11 @@ class TestDataLoadingForPartitionTable extends QueryTest with BeforeAndAfterAll
sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE originMultiLoads OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '"')""")
}
- def validateDataFiles(tableUniqueName: String, sgementId: String, partitions: Seq[Int]): Unit = {
+ def validateDataFiles(tableUniqueName: String, segmentId: String, partitions: Seq[Int]): Unit = {
val carbonTable = CarbonMetadata.getInstance().getCarbonTable(tableUniqueName)
val tablePath = new CarbonTablePath(carbonTable.getStorePath, carbonTable.getDatabaseName,
carbonTable.getFactTableName)
- val segmentDir = tablePath.getCarbonDataDirectoryPath("0", sgementId)
+ val segmentDir = tablePath.getCarbonDataDirectoryPath("0", segmentId)
val carbonFile = FileFactory.getCarbonFile(segmentDir, FileFactory.getFileType(segmentDir))
val dataFiles = carbonFile.listFiles(new CarbonFileFilter() {
override def accept(file: CarbonFile): Boolean = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/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 905f977..24c09ca 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
@@ -209,12 +209,12 @@ public final class CarbonLoaderUtil {
* @param loadModel
*/
public static void deleteLocalDataLoadFolderLocation(CarbonLoadModel loadModel,
- boolean isCompactionFlow) {
+ boolean isCompactionFlow, boolean isAltPartitionFlow) {
String databaseName = loadModel.getDatabaseName();
String tableName = loadModel.getTableName();
String tempLocationKey = CarbonDataProcessorUtil
.getTempStoreLocationKey(databaseName, tableName, loadModel.getSegmentId(),
- loadModel.getTaskNo(), isCompactionFlow);
+ loadModel.getTaskNo(), isCompactionFlow, isAltPartitionFlow);
// form local store location
final String localStoreLocations = CarbonProperties.getInstance().getProperty(tempLocationKey);
if (localStoreLocations == null) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/SplitPartitionCallable.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/SplitPartitionCallable.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/SplitPartitionCallable.java
new file mode 100644
index 0000000..3a37a9d
--- /dev/null
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/partition/SplitPartitionCallable.java
@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+import java.util.concurrent.Callable;
+
+import org.apache.carbondata.spark.rdd.PartitionSplitter;
+
+import org.apache.spark.sql.execution.command.SplitPartitionCallableModel;
+
+/**
+ * Callable class which is used to split the partition in a separate callable.
+ */
+public class SplitPartitionCallable implements Callable<Void> {
+
+ private final SplitPartitionCallableModel splitPartitionCallableModel;
+
+ public SplitPartitionCallable(SplitPartitionCallableModel splitPartitionCallableModel) {
+ this.splitPartitionCallableModel = splitPartitionCallableModel;
+ }
+
+ @Override public Void call() {
+ PartitionSplitter.triggerPartitionSplit(splitPartitionCallableModel);
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/KeyVal.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/KeyVal.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/KeyVal.scala
index 31dd4e6..181f6e4 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/KeyVal.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/KeyVal.scala
@@ -107,6 +107,14 @@ class MergeResultImpl extends MergeResult[String, Boolean] {
override def getKey(key: String, value: Boolean): (String, Boolean) = (key, value)
}
+trait SplitResult[K, V] extends Serializable {
+ def getKey(key: String, value: Boolean): (K, V)
+}
+
+class SplitResultImpl extends SplitResult[String, Boolean] {
+ override def getKey(key: String, value: Boolean): (String, Boolean) = (key, value)
+}
+
trait DeletedLoadResult[K, V] extends Serializable {
def getKey(key: String, value: String): (K, V)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessorStepOnSpark.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessorStepOnSpark.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessorStepOnSpark.scala
index 4d1267a..e43d204 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessorStepOnSpark.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/load/DataLoadProcessorStepOnSpark.scala
@@ -204,7 +204,7 @@ object DataLoadProcessorStepOnSpark {
dataWriter.close()
}
// clean up the folders and files created locally for data load operation
- CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false)
+ CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false, false)
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableSplitPartitionRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableSplitPartitionRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableSplitPartitionRDD.scala
new file mode 100644
index 0000000..e481fc4
--- /dev/null
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/AlterTableSplitPartitionRDD.scala
@@ -0,0 +1,146 @@
+/*
+ * 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.rdd
+
+import scala.collection.JavaConverters._
+import scala.util.Random
+
+import org.apache.spark.{Partition, SparkContext, SparkEnv, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.util.PartitionUtils
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
+import org.apache.carbondata.core.metadata.schema.PartitionInfo
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.processing.model.CarbonLoadModel
+import org.apache.carbondata.processing.spliter.RowResultSpliterProcessor
+import org.apache.carbondata.processing.util.CarbonDataProcessorUtil
+import org.apache.carbondata.spark.SplitResult
+import org.apache.carbondata.spark.load.CarbonLoaderUtil
+
+class AlterTableSplitPartitionRDD[K, V](
+ sc: SparkContext,
+ result: SplitResult[K, V],
+ partitionIds: Seq[String],
+ segmentId: String,
+ bucketId: Int,
+ carbonLoadModel: CarbonLoadModel,
+ identifier: AbsoluteTableIdentifier,
+ storePath: String,
+ oldPartitionIdList: List[Int],
+ prev: RDD[Array[AnyRef]]) extends RDD[(K, V)](prev) {
+
+ sc.setLocalProperty("spark.scheduler.pool", "DDL")
+ sc.setLocalProperty("spark.job.interruptOnCancel", "true")
+
+ var storeLocation: String = null
+ var splitResult: String = null
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val databaseName = carbonTable.getDatabaseName
+ val factTableName = carbonTable.getFactTableName
+ val partitionInfo = carbonTable.getPartitionInfo(factTableName)
+
+ override protected def getPartitions: Array[Partition] = firstParent[Array[AnyRef]].partitions
+
+ override def compute(split: Partition, context: TaskContext): Iterator[(K, V)] = {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+ val rows = firstParent[Array[AnyRef]].iterator(split, context).toList.asJava
+ val iter = new Iterator[(K, V)] {
+ val partitionId = partitionInfo.getPartitionId(split.index)
+ carbonLoadModel.setTaskNo(String.valueOf(partitionId))
+ carbonLoadModel.setSegmentId(segmentId)
+ carbonLoadModel.setPartitionId("0")
+ val tempLocationKey = CarbonDataProcessorUtil
+ .getTempStoreLocationKey(carbonLoadModel.getDatabaseName,
+ carbonLoadModel.getTableName,
+ segmentId,
+ carbonLoadModel.getTaskNo,
+ false,
+ true)
+ // 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")) {
+
+ val storeLocations = CarbonLoaderUtil.getConfiguredLocalDirs(SparkEnv.get.conf)
+ if (null != storeLocations && storeLocations.nonEmpty) {
+ storeLocation = storeLocations(Random.nextInt(storeLocations.length))
+ }
+ if (storeLocation == null) {
+ storeLocation = System.getProperty("java.io.tmpdir")
+ }
+ } else {
+ storeLocation = System.getProperty("java.io.tmpdir")
+ }
+ storeLocation = storeLocation + '/' + System.nanoTime() + '/' + split.index
+ CarbonProperties.getInstance().addProperty(tempLocationKey, storeLocation)
+ LOGGER.info(s"Temp storeLocation taken is $storeLocation")
+
+ val tempStoreLoc = CarbonDataProcessorUtil.getLocalDataFolderLocation(databaseName,
+ factTableName,
+ carbonLoadModel.getTaskNo,
+ "0",
+ segmentId,
+ false,
+ true
+ )
+
+ val splitStatus = if (rows.isEmpty) {
+ LOGGER.info("After repartition this split, NO target rows to write back.")
+ true
+ } else {
+ try {
+ val segmentProperties = PartitionUtils.getSegmentProperties(identifier,
+ segmentId, partitionIds.toList, oldPartitionIdList, partitionInfo)
+ val processor = new RowResultSpliterProcessor(
+ carbonTable,
+ carbonLoadModel,
+ segmentProperties,
+ tempStoreLoc,
+ bucketId
+ )
+ processor.execute(rows)
+ } catch {
+ case e: Exception =>
+ sys.error(s"Exception when executing Row result processor ${e.getMessage}")
+ } finally {
+ CarbonLoaderUtil
+ .deleteLocalDataLoadFolderLocation(carbonLoadModel, false, true)
+ }
+
+ }
+
+ val splitResult = segmentId
+ var finished = false
+
+ override def hasNext: Boolean = {
+ !finished
+ }
+
+ override def next(): (K, V) = {
+ finished = true
+ result.getKey(splitResult, splitStatus)
+ }
+ }
+ iter
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/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 add0578..07264b5 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
@@ -153,7 +153,8 @@ class CarbonMergerRDD[K, V](
carbonLoadModel.getTableName,
carbonLoadModel.getSegmentId,
carbonLoadModel.getTaskNo,
- true)
+ true,
+ false)
CarbonProperties.getInstance().addProperty(tempLocationKey, storeLocation)
LOGGER.info(s"Temp storeLocation taken is $storeLocation")
// get destination segment properties as sent from driver which is of last segment.
@@ -196,7 +197,8 @@ class CarbonMergerRDD[K, V](
carbonLoadModel.getTaskNo,
"0",
mergeNumber,
- true
+ true,
+ false
)
carbonLoadModel.setPartitionId("0")
@@ -231,7 +233,7 @@ class CarbonMergerRDD[K, V](
try {
val isCompactionFlow = true
CarbonLoaderUtil
- .deleteLocalDataLoadFolderLocation(carbonLoadModel, isCompactionFlow)
+ .deleteLocalDataLoadFolderLocation(carbonLoadModel, isCompactionFlow, false)
} catch {
case e: Exception =>
LOGGER.error(e)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
new file mode 100644
index 0000000..2a39db5
--- /dev/null
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanPartitionRDD.scala
@@ -0,0 +1,269 @@
+/*
+ * 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.rdd
+
+import java.util.ArrayList
+
+import scala.collection.JavaConverters._
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.mapred.JobConf
+import org.apache.hadoop.mapreduce.Job
+import org.apache.spark.{Partition, SparkContext, TaskContext}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.hive.DistributionUtil
+import org.apache.spark.unsafe.types.UTF8String
+import org.apache.spark.util.PartitionUtils
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.datastore.block.{Distributable, SegmentProperties, TaskBlockInfo}
+import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory
+import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
+import org.apache.carbondata.core.metadata.datatype.DataType
+import org.apache.carbondata.core.metadata.encoder.Encoding
+import org.apache.carbondata.core.metadata.schema.partition.PartitionType
+import org.apache.carbondata.core.metadata.schema.table.column.{CarbonDimension, CarbonMeasure}
+import org.apache.carbondata.core.scan.result.iterator.PartitionSpliterRawResultIterator
+import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper
+import org.apache.carbondata.core.util.{ByteUtil, DataTypeUtil}
+import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonMultiBlockSplit}
+import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
+import org.apache.carbondata.processing.merger.CarbonCompactionUtil
+import org.apache.carbondata.processing.model.CarbonLoadModel
+import org.apache.carbondata.processing.spliter.CarbonSplitExecutor
+import org.apache.carbondata.spark.load.CarbonLoaderUtil
+
+
+/**
+ * This RDD is used in alter table partition statement to get data of target partitions,
+ * then repartition data according to new partitionInfo
+ * @param sc
+ * @param partitionIds the ids of target partition to be scanned
+ * @param storePath
+ * @param segmentId
+ * @param bucketId
+ * @param oldPartitionIdList the taskId in partition order before partitionInfo is modified
+ * @param carbonTableIdentifier
+ * @param carbonLoadModel
+ */
+class CarbonScanPartitionRDD(
+ sc: SparkContext,
+ partitionIds: Seq[String],
+ storePath: String,
+ segmentId: String,
+ bucketId: Int,
+ oldPartitionIdList: List[Int],
+ carbonTableIdentifier: CarbonTableIdentifier,
+ carbonLoadModel: CarbonLoadModel)
+ extends RDD[(AnyRef, Array[AnyRef])](sc, Nil) {
+
+ private val queryId = sc.getConf.get("queryId", System.nanoTime() + "")
+ val identifier = new AbsoluteTableIdentifier(storePath, carbonTableIdentifier)
+ var storeLocation: String = null
+ var splitStatus: Boolean = false
+ var blockId: String = null
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val dimensions = carbonTable.getAllDimensions.asScala
+ val measures = carbonTable.getAllMeasures.asScala
+ val partitionInfo = carbonTable.getPartitionInfo(carbonTableIdentifier.getTableName)
+ val partitionColumn = partitionInfo.getColumnSchemaList().get(0)
+ val partitionDataType = partitionColumn.getDataType
+ val partitionColumnName = partitionColumn.getColumnName
+ var isDimension: Boolean = false
+ val encodingList = partitionColumn.getEncodingList
+ var dimension: CarbonDimension = null
+ var measure: CarbonMeasure = null
+ val noDictionaryIndexGroup: ArrayBuffer[Int] = new ArrayBuffer[Int]()
+ val dictionaryIndexGroup: ArrayBuffer[Int] = new ArrayBuffer[Int]()
+ val measureIndexGroup: ArrayBuffer[Int] = new ArrayBuffer[Int]()
+
+ override def getPartitions: Array[Partition] = {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+ val parallelism = sparkContext.defaultParallelism
+ val jobConf = new JobConf(new Configuration)
+ val job = new Job(jobConf)
+ val format = CarbonInputFormatUtil.createCarbonTableInputFormat(identifier,
+ partitionIds.toList.asJava, job)
+ job.getConfiguration.set("query.id", queryId)
+
+ val splits = format.getSplitsOfOneSegment(job, segmentId,
+ oldPartitionIdList.map(_.asInstanceOf[Integer]).asJava, partitionInfo)
+ var partition_num = 0
+ val result = new ArrayList[Partition](parallelism)
+ val blockList = splits.asScala
+ .filter(_.asInstanceOf[CarbonInputSplit].getBucketId.toInt == bucketId)
+ .map(_.asInstanceOf[Distributable])
+ if (!blockList.isEmpty) {
+ val activeNodes = DistributionUtil.ensureExecutorsAndGetNodeList(blockList, sparkContext)
+ val nodeBlockMapping = CarbonLoaderUtil.nodeBlockTaskMapping(blockList.asJava, -1,
+ parallelism, activeNodes.toList.asJava)
+ nodeBlockMapping.asScala.foreach { case (node, blockList) =>
+ blockList.asScala.foreach { blocksPerTask =>
+ val splits = blocksPerTask.asScala.map(_.asInstanceOf[CarbonInputSplit])
+ if (blocksPerTask.size() != 0) {
+ val multiBlockSplit =
+ new CarbonMultiBlockSplit(identifier, splits.asJava, Array(node))
+ val partition = new CarbonSparkPartition(id, partition_num, multiBlockSplit)
+ result.add(partition)
+ partition_num += 1
+ }
+ }
+ }
+ }
+ result.toArray(new Array[Partition](result.size()))
+ }
+
+ override def compute(split: Partition, context: TaskContext):
+ Iterator[(AnyRef, Array[AnyRef])] = {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)
+ var exec : CarbonSplitExecutor = null
+ val rows : java.util.List[(AnyRef, Array[AnyRef])] = new ArrayList[(AnyRef, Array[AnyRef])]()
+ try {
+ val inputSplit = split.asInstanceOf[CarbonSparkPartition].split.value
+ val splits = inputSplit.getAllSplits.asScala
+ val tableBlockInfoList = CarbonInputSplit.createBlocks(splits.asJava)
+ val segmentMapping: java.util.Map[String, TaskBlockInfo] =
+ CarbonCompactionUtil.createMappingForSegments(tableBlockInfoList)
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ var result : java.util.List[PartitionSpliterRawResultIterator] = null
+ try {
+ exec = new CarbonSplitExecutor(segmentMapping, carbonTable)
+ result = exec.processDataBlocks(segmentId)
+ } catch {
+ case e: Throwable =>
+ LOGGER.error(e)
+ if (null != e.getMessage) {
+ sys.error(s"Exception occurred in query execution :: ${ e.getMessage }")
+ } else {
+ sys.error("Exception occurred in query execution. Please check logs.")
+ }
+ }
+ val segmentProperties = PartitionUtils.getSegmentProperties(identifier, segmentId,
+ partitionIds.toList, oldPartitionIdList, partitionInfo)
+ val partColIdx = getPartitionColumnIndex(partitionColumnName, segmentProperties)
+ indexInitialise()
+ for (iterator <- result.asScala) {
+ while (iterator.hasNext) {
+ val row = iterator.next()
+ val partitionColumnValue = getPartitionColumnValue(row, partColIdx,
+ segmentProperties)
+ rows.add((partitionColumnValue, row))
+ }
+ }
+ } catch {
+ case e: Exception =>
+ LOGGER.error(e)
+ throw e
+ } finally {
+ if (null != exec) {
+ exec.finish
+ }
+ }
+ val iter = rows.iterator().asScala
+ iter
+ }
+
+ def getPartitionColumnValue(row: Array[AnyRef], partColIdx: Int,
+ segmentProperties: SegmentProperties): AnyRef = {
+ val dims: Array[Byte] = row(0).asInstanceOf[ByteArrayWrapper].getDictionaryKey
+ val keyGen = segmentProperties.getDimensionKeyGenerator
+ val keyArray: Array[Long] = keyGen.getKeyArray(dims)
+ val encodings = partitionColumn.getEncodingList
+ val partitionType = partitionInfo.getPartitionType
+ var partitionValue: AnyRef = null
+ val factor = 1000L
+ if (isDimension) {
+ // direct dictionary
+ if (encodings.contains(Encoding.DIRECT_DICTIONARY)) {
+ val directDictionaryGenerator = DirectDictionaryKeyGeneratorFactory
+ .getDirectDictionaryGenerator(partitionDataType)
+ val dictionaryIndex = dictionaryIndexGroup.indexOf(partColIdx)
+ val surrogateValue = (keyArray(dictionaryIndex) / factor).toInt
+ partitionValue = directDictionaryGenerator.getValueFromSurrogate(surrogateValue)
+ } else if (!encodings.contains(Encoding.DICTIONARY)) {
+ // no dictionary
+ val byteArray = row(0).asInstanceOf[ByteArrayWrapper].getNoDictionaryKeys
+ val index = noDictionaryIndexGroup.indexOf(partColIdx)
+ partitionValue = DataTypeUtil.getDataBasedOnDataTypeForNoDictionaryColumn(byteArray(index)
+ , partitionDataType)
+ if (partitionValue.isInstanceOf[UTF8String]) {
+ partitionValue = partitionValue.toString
+ }
+ } else { // normal dictionary
+ val dict = CarbonLoaderUtil.getDictionary(carbonTableIdentifier,
+ dimension.getColumnIdentifier, storePath, partitionDataType)
+ if (partitionDataType == DataType.STRING) {
+ if (partitionType == PartitionType.RANGE) {
+ partitionValue = ByteUtil.
+ toBytes(dict.getDictionaryValueForKey(keyArray(partColIdx).toInt))
+ } else {
+ partitionValue = dict.getDictionaryValueForKey(keyArray(partColIdx).toInt)
+ }
+ } else {
+ partitionValue = dict.getDictionaryValueForKey(keyArray(partColIdx).toInt)
+ }
+
+ }
+ } else {
+ partitionValue = row(measureIndexGroup(partColIdx))
+ }
+ partitionValue
+ }
+
+ def indexInitialise(): Unit = {
+ for (dim: CarbonDimension <- dimensions) {
+ if (!dim.getEncoder.contains(Encoding.DICTIONARY)) {
+ noDictionaryIndexGroup.append(dimensions.indexOf(dim))
+ } else {
+ dictionaryIndexGroup.append(dimensions.indexOf(dim))
+ }
+ }
+ for (msr: CarbonMeasure <- measures) {
+ // index of measure in row
+ measureIndexGroup.append(measures.indexOf(msr) + 1)
+ }
+ }
+
+ /**
+ * get the index of partition column in dimension/measure
+ * @param partitionColumnName
+ * @param segmentProperties
+ * @return
+ */
+ def getPartitionColumnIndex(partitionColumnName: String,
+ segmentProperties: SegmentProperties): Int = {
+ val dimensions = segmentProperties.getDimensions
+ val measures = segmentProperties.getMeasures
+ val columns = dimensions.asScala.map(_.getColName) ++ measures.asScala.map(_.getColName)
+ var index = 0
+ for (i <- 0 until columns.size) {
+ if (columns(i) == partitionColumnName) {
+ index = i
+ }
+ }
+ if (index < dimensions.size()) {
+ isDimension = true
+ dimension = dimensions.get(index)
+ } else {
+ index = index - dimensions.size()
+ measure = measures.get(index)
+ }
+ index
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
index a6a8835..bca119e 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/DataManagementFunc.scala
@@ -24,7 +24,7 @@ import scala.collection.JavaConverters._
import scala.collection.mutable.ListBuffer
import org.apache.spark.sql.SQLContext
-import org.apache.spark.sql.execution.command.{CompactionCallableModel, CompactionModel}
+import org.apache.spark.sql.execution.command.{CompactionCallableModel, CompactionModel, SplitPartitionCallableModel}
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
@@ -39,6 +39,7 @@ import org.apache.carbondata.processing.model.{CarbonDataLoadSchema, CarbonLoadM
import org.apache.carbondata.spark._
import org.apache.carbondata.spark.compaction.CompactionCallable
import org.apache.carbondata.spark.load._
+import org.apache.carbondata.spark.partition.SplitPartitionCallable
import org.apache.carbondata.spark.util.{CommonUtil, LoadMetadataUtil}
/**
@@ -260,6 +261,50 @@ object DataManagementFunc {
futureList.add(future)
}
+ def executePartitionSplit( sqlContext: SQLContext,
+ carbonLoadModel: CarbonLoadModel,
+ executor: ExecutorService,
+ storePath: String,
+ segment: String,
+ partitionId: String,
+ oldPartitionIdList: List[Int]): Unit = {
+ val futureList: util.List[Future[Void]] = new util.ArrayList[Future[Void]](
+ CarbonCommonConstants.DEFAULT_COLLECTION_SIZE
+ )
+ scanSegmentsForSplitPartition(futureList, executor, storePath, segment, partitionId,
+ sqlContext, carbonLoadModel, oldPartitionIdList)
+ try {
+ futureList.asScala.foreach(future => {
+ future.get
+ }
+ )
+ } catch {
+ case e: Exception =>
+ LOGGER.error(e, s"Exception in partition split thread ${ e.getMessage }")
+ throw e
+ }
+ }
+
+ private def scanSegmentsForSplitPartition(futureList: util.List[Future[Void]],
+ executor: ExecutorService,
+ storePath: String,
+ segmentId: String,
+ partitionId: String,
+ sqlContext: SQLContext,
+ carbonLoadModel: CarbonLoadModel,
+ oldPartitionIdList: List[Int]): Unit = {
+
+ val splitModel = SplitPartitionCallableModel(storePath,
+ carbonLoadModel,
+ segmentId,
+ partitionId,
+ oldPartitionIdList,
+ sqlContext)
+
+ val future: Future[Void] = executor.submit(new SplitPartitionCallable(splitModel))
+ futureList.add(future)
+ }
+
def prepareCarbonLoadModel(storePath: String,
table: CarbonTable,
newCarbonLoadModel: CarbonLoadModel): Unit = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/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 b51800e..37b2d02 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
@@ -267,7 +267,7 @@ class NewCarbonDataLoadRDD[K, V](
throw e
} finally {
// clean up the folders and files created locally for data load operation
- CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false)
+ CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false, false)
// in case of failure the same operation will be re-tried several times.
// So print the data load statistics only in case of non failure case
if (!CarbonCommonConstants.STORE_LOADSTATUS_FAILURE
@@ -453,7 +453,7 @@ class NewDataFrameLoaderRDD[K, V](
throw e
} finally {
// clean up the folders and files created locally for data load operation
- CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false)
+ CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false, false)
// in case of failure the same operation will be re-tried several times.
// So print the data load statistics only in case of non failure case
if (!CarbonCommonConstants.STORE_LOADSTATUS_FAILURE
@@ -638,7 +638,7 @@ class PartitionTableDataLoaderRDD[K, V](
throw e
} finally {
// clean up the folders and files created locally for data load operation
- CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false)
+ CarbonLoaderUtil.deleteLocalDataLoadFolderLocation(model, false, false)
// in case of failure the same operation will be re-tried several times.
// So print the data load statistics only in case of non failure case
if (!CarbonCommonConstants.STORE_LOADSTATUS_FAILURE
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
new file mode 100644
index 0000000..48e1bee
--- /dev/null
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/PartitionSplitter.scala
@@ -0,0 +1,104 @@
+/*
+ * 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.rdd
+
+import java.io.IOException
+
+import org.apache.spark.sql.execution.command.SplitPartitionCallableModel
+import org.apache.spark.util.PartitionUtils
+
+import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.spark.{PartitionFactory, SplitResultImpl}
+
+object PartitionSplitter {
+
+ val logger = LogServiceFactory.getLogService(PartitionSplitter.getClass.getName)
+
+ def triggerPartitionSplit(splitPartitionCallableModel: SplitPartitionCallableModel): Unit = {
+ val sc = splitPartitionCallableModel.sqlContext.sparkContext
+ val partitionId = splitPartitionCallableModel.partitionId
+ val storePath = splitPartitionCallableModel.storePath
+ val segmentId = splitPartitionCallableModel.segmentId
+ val oldPartitionIdList = splitPartitionCallableModel.oldPartitionIdList
+ val carbonLoadModel = splitPartitionCallableModel.carbonLoadModel
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ val identifier = carbonTable.getAbsoluteTableIdentifier
+ val carbonTableIdentifier = identifier.getCarbonTableIdentifier
+ val tableName = carbonTable.getFactTableName
+ val databaseName = carbonTable.getDatabaseName
+ val bucketInfo = carbonTable.getBucketingInfo(tableName)
+ var finalSplitStatus = false
+ val bucketNumber = bucketInfo match {
+ case null => 1
+ case _ => bucketInfo.getNumberOfBuckets
+ }
+ val partitionInfo = carbonTable.getPartitionInfo(tableName)
+ val partitioner = PartitionFactory.getPartitioner(partitionInfo)
+
+ for (i <- 0 until bucketNumber) {
+ val bucketId = i
+ val rdd = new CarbonScanPartitionRDD(
+ sc,
+ Seq(partitionId),
+ storePath,
+ segmentId,
+ bucketId,
+ oldPartitionIdList,
+ carbonTableIdentifier,
+ carbonLoadModel
+ ).partitionBy(partitioner).map(_._2)
+
+ val splitStatus = new AlterTableSplitPartitionRDD(sc,
+ new SplitResultImpl(),
+ Seq(partitionId),
+ segmentId,
+ bucketId,
+ carbonLoadModel,
+ identifier,
+ storePath,
+ oldPartitionIdList,
+ rdd).collect()
+
+ if (splitStatus.length == 0) {
+ finalSplitStatus = false
+ } else {
+ finalSplitStatus = splitStatus.forall(_._2)
+ }
+ if (!finalSplitStatus) {
+ logger.audit(s"Add/Split Partition request failed for table " +
+ s"${ databaseName }.${ tableName }")
+ logger.error(s"Add/Split Partition request failed for table " +
+ s"${ databaseName }.${ tableName }")
+ }
+ }
+ if (finalSplitStatus) {
+ try {
+ PartitionUtils.
+ deleteOriginalCarbonFile(identifier, segmentId, Seq(partitionId).toList,
+ oldPartitionIdList, storePath, databaseName, tableName, partitionInfo, carbonLoadModel)
+ } catch {
+ case e: IOException => sys.error(s"Exception while delete original carbon files " +
+ e.getMessage)
+ }
+ logger.audit(s"Add/Split Partition request completed for table " +
+ s"${ databaseName }.${ tableName }")
+ logger.info(s"Add/Split Partition request completed for table " +
+ s"${ databaseName }.${ tableName }")
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index 705daea..4d781a1 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -292,11 +292,13 @@ object CommonUtil {
value.matches(pattern)
case "timestamp" =>
val timeStampFormat = new SimpleDateFormat(CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT))
+ .getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
+ CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT))
scala.util.Try(timeStampFormat.parse(value)).isSuccess
case "date" =>
val dateFormat = new SimpleDateFormat(CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.CARBON_DATE_FORMAT))
+ .getProperty(CarbonCommonConstants.CARBON_DATE_FORMAT,
+ CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT))
scala.util.Try(dateFormat.parse(value)).isSuccess
case _ =>
validateTypeConvertForSpark2(partitionerField, value)
@@ -333,6 +335,40 @@ object CommonUtil {
}
}
+ def validateSplitListInfo(originListInfo: List[String], newListInfo: List[String],
+ originList: List[List[String]]): Unit = {
+ if (originListInfo.size == 1) {
+ sys.error("The target list partition cannot be split, please check again!")
+ }
+ if (newListInfo.size == 1) {
+ sys.error("Can't split list to one partition, please check again!")
+ }
+ if (!(newListInfo.size < originListInfo.size)) {
+ sys.error("The size of new list must be smaller than original list, please check again!")
+ }
+ val tempList = newListInfo.mkString(",").split(",")
+ .map(_.trim.replace("(", "").replace(")", ""))
+ if (tempList.length != originListInfo.size) {
+ sys.error("The total number of elements in new list must equal to original list!")
+ }
+ if (!originListInfo.sameElements(tempList)) {
+ sys.error("The elements in new list must exist in original list")
+ }
+ }
+
+ def validateAddListInfo(newListInfo: List[String], originList: List[List[String]]): Unit = {
+ if (newListInfo.size < 1) {
+ sys.error("Please add at least one new partition")
+ }
+ for (originElementGroup <- originList) {
+ for (newElement <- newListInfo ) {
+ if (originElementGroup.contains(newElement)) {
+ sys.error(s"The partition $newElement is already exist! Please check again!")
+ }
+ }
+ }
+ }
+
def validateFields(key: String, fields: Seq[Field]): Boolean = {
var isValid: Boolean = false
fields.foreach { field =>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/874764f9/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
index 500e18e..70a2498 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/execution/command/carbonTableSchema.scala
@@ -126,6 +126,13 @@ case class CompactionCallableModel(storePath: String,
sqlContext: SQLContext,
compactionType: CompactionType)
+case class SplitPartitionCallableModel(storePath: String,
+ carbonLoadModel: CarbonLoadModel,
+ segmentId: String,
+ partitionId: String,
+ oldPartitionIdList: List[Int],
+ sqlContext: SQLContext)
+
case class DataTypeInfo(dataType: String, precision: Int = 0, scale: Int = 0)
case class AlterTableDataTypeChangeModel(dataTypeInfo: DataTypeInfo,
@@ -151,6 +158,15 @@ case class AlterTableDropColumnModel(databaseName: Option[String],
tableName: String,
columns: List[String])
+case class AlterTableDropPartitionModel(databaseName: Option[String],
+ tableName: String,
+ partitionId: String)
+
+case class AlterTableSplitPartitionModel(databaseName: Option[String],
+ tableName: String,
+ partitionId: String,
+ splitInfo: List[String])
+
class AlterTableColumnSchemaGenerator(
alterTableModel: AlterTableAddColumnsModel,
dbName: String,