You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ma...@apache.org on 2018/04/05 09:59:15 UTC
[1/2] carbondata git commit: [CARBONDATA-2294] Partition preaggregate
support
Repository: carbondata
Updated Branches:
refs/heads/master 248e0c850 -> 550846029
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
index 8740a2e..ce4e6fa 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
@@ -26,9 +26,11 @@ import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.AlterTableModel
import org.apache.spark.sql.execution.command.management.{CarbonAlterTableCompactionCommand, CarbonLoadDataCommand}
+import org.apache.spark.sql.execution.command.partition.CarbonAlterTableDropHivePartitionCommand
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
-import org.apache.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.common.exceptions.MetadataProcessException
+import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
import org.apache.carbondata.core.datastore.impl.FileFactory
@@ -39,6 +41,216 @@ import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events._
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent, LoadTablePostStatusUpdateEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
+object AlterTableDropPartitionPreStatusListener extends OperationEventListener {
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override protected def onEvent(event: Event,
+ operationContext: OperationContext) = {
+ val preStatusListener = event.asInstanceOf[AlterTableDropPartitionPreStatusEvent]
+ val carbonTable = preStatusListener.carbonTable
+ val childDropPartitionCommands = operationContext.getProperty("dropPartitionCommands")
+ if (childDropPartitionCommands != null && carbonTable.hasAggregationDataMap) {
+ val childCommands =
+ childDropPartitionCommands.asInstanceOf[Seq[CarbonAlterTableDropHivePartitionCommand]]
+ childCommands.foreach(_.processData(SparkSession.getActiveSession.get))
+ }
+ }
+}
+
+trait CommitHelper {
+
+ val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getName)
+
+ protected def markInProgressSegmentAsDeleted(tableStatusFile: String,
+ operationContext: OperationContext,
+ carbonTable: CarbonTable): Unit = {
+ val loadMetaDataDetails = SegmentStatusManager.readTableStatusFile(tableStatusFile)
+ val segmentBeingLoaded =
+ operationContext.getProperty(carbonTable.getTableUniqueName + "_Segment").toString
+ val newDetails = loadMetaDataDetails.collect {
+ case detail if detail.getLoadName.equalsIgnoreCase(segmentBeingLoaded) =>
+ detail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE)
+ detail
+ case others => others
+ }
+ SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusFile, newDetails)
+ }
+
+ /**
+ * Used to rename table status files for commit operation.
+ */
+ protected def renameDataMapTableStatusFiles(sourceFileName: String,
+ destinationFileName: String, uuid: String): Boolean = {
+ val oldCarbonFile = FileFactory.getCarbonFile(sourceFileName)
+ val newCarbonFile = FileFactory.getCarbonFile(destinationFileName)
+ if (oldCarbonFile.exists() && newCarbonFile.exists()) {
+ val backUpPostFix = if (uuid.nonEmpty) {
+ "_backup_" + uuid
+ } else {
+ ""
+ }
+ LOGGER.info(s"Renaming $newCarbonFile to ${destinationFileName + backUpPostFix}")
+ if (newCarbonFile.renameForce(destinationFileName + backUpPostFix)) {
+ LOGGER.info(s"Renaming $oldCarbonFile to $destinationFileName")
+ oldCarbonFile.renameForce(destinationFileName)
+ } else {
+ LOGGER.info(s"Renaming $newCarbonFile to ${destinationFileName + backUpPostFix} failed")
+ false
+ }
+ } else {
+ false
+ }
+ }
+
+ /**
+ * Used to remove table status files with UUID and segment folders.
+ */
+ protected def cleanUpStaleTableStatusFiles(
+ childTables: Seq[CarbonTable],
+ operationContext: OperationContext,
+ uuid: String): Unit = {
+ childTables.foreach { childTable =>
+ val metaDataDir = FileFactory.getCarbonFile(
+ CarbonTablePath.getMetadataPath(childTable.getTablePath))
+ val tableStatusFiles = metaDataDir.listFiles(new CarbonFileFilter {
+ override def accept(file: CarbonFile): Boolean = {
+ file.getName.contains(uuid) || file.getName.contains("backup")
+ }
+ })
+ tableStatusFiles.foreach(_.delete())
+ }
+ }
+}
+
+object AlterTableDropPartitionPostStatusListener extends OperationEventListener with CommitHelper {
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override protected def onEvent(event: Event,
+ operationContext: OperationContext) = {
+ val postStatusListener = event.asInstanceOf[AlterTableDropPartitionPostStatusEvent]
+ val carbonTable = postStatusListener.carbonTable
+ val childDropPartitionCommands = operationContext.getProperty("dropPartitionCommands")
+ val uuid = Option(operationContext.getProperty("uuid")).getOrElse("").toString
+ if (childDropPartitionCommands != null && carbonTable.hasAggregationDataMap) {
+ val childCommands =
+ childDropPartitionCommands.asInstanceOf[Seq[CarbonAlterTableDropHivePartitionCommand]]
+ val updateFailed = try {
+ val renamedDataMaps = childCommands.takeWhile {
+ childCommand =>
+ val childCarbonTable = childCommand.table
+ val oldTableSchemaPath = CarbonTablePath.getTableStatusFilePathWithUUID(
+ childCarbonTable.getTablePath, uuid)
+ // Generate table status file name without UUID, forExample: tablestatus
+ val newTableSchemaPath = CarbonTablePath.getTableStatusFilePath(
+ childCarbonTable.getTablePath)
+ renameDataMapTableStatusFiles(oldTableSchemaPath, newTableSchemaPath, uuid)
+ }
+ // if true then the commit for one of the child tables has failed
+ val commitFailed = renamedDataMaps.lengthCompare(childCommands.length) != 0
+ if (commitFailed) {
+ LOGGER.info("Reverting table status file to original state")
+ renamedDataMaps.foreach {
+ command =>
+ val carbonTable = command.table
+ // rename the backup tablestatus i.e tablestatus_backup_UUID to tablestatus
+ val backupTableSchemaPath =
+ CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath) + "_backup_" + uuid
+ val tableSchemaPath = CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath)
+ renameDataMapTableStatusFiles(backupTableSchemaPath, tableSchemaPath, "")
+ }
+ }
+ commitFailed
+ } finally {
+ // after success/failure of commit delete all tablestatus files with UUID in their names.
+ // if commit failed then remove the segment directory
+ // TODO Need to handle deletion on tablestatus files with UUID in cleanup command.
+ cleanUpStaleTableStatusFiles(childCommands.map(_.table),
+ operationContext,
+ uuid)
+ }
+ if (updateFailed) {
+ sys.error("Failed to update table status for pre-aggregate table")
+ }
+ }
+ }
+}
+
+object AlterTableDropPartitionMetaListener extends OperationEventListener{
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override protected def onEvent(event: Event,
+ operationContext: OperationContext) = {
+ val dropPartitionEvent = event.asInstanceOf[AlterTableDropPartitionMetaEvent]
+ val parentCarbonTable = dropPartitionEvent.parentCarbonTable
+ val partitionsToBeDropped = dropPartitionEvent.specs.flatMap(_.keys)
+ val sparkSession = SparkSession.getActiveSession.get
+ if (parentCarbonTable.hasAggregationDataMap) {
+ // used as a flag to block direct drop partition on aggregate tables fired by the user
+ operationContext.setProperty("isInternalDropCall", "true")
+ // Filter out all the tables which dont have the partition being dropped.
+ val childTablesWithoutPartitionColumns =
+ parentCarbonTable.getTableInfo.getDataMapSchemaList.asScala.filter { dataMapSchema =>
+ val childColumns = dataMapSchema.getChildSchema.getListOfColumns.asScala
+ val partitionColExists = partitionsToBeDropped.forall {
+ partition =>
+ childColumns.exists { childColumn =>
+ childColumn.getAggFunction.isEmpty &&
+ childColumn.getParentColumnTableRelations.asScala.head.getColumnName.
+ equals(partition)
+ }
+ }
+ !partitionColExists
+ }
+ if (childTablesWithoutPartitionColumns.nonEmpty) {
+ throw new MetadataProcessException(s"Cannot drop partition as one of the partition is not" +
+ s" participating in the following datamaps ${
+ childTablesWithoutPartitionColumns.toList.map(_.getChildSchema.getTableName)
+ }. Please drop the specified aggregate tables to continue")
+ } else {
+ val childDropPartitionCommands =
+ parentCarbonTable.getTableInfo.getDataMapSchemaList.asScala.map { dataMapSchema =>
+ val tableIdentifier = TableIdentifier(dataMapSchema.getChildSchema.getTableName,
+ Some(parentCarbonTable.getDatabaseName))
+ // as the aggregate table columns start with parent table name therefore the
+ // partition column also has to be updated with parent table name to generate
+ // partitionSpecs for the child table.
+ val childSpecs = dropPartitionEvent.specs.map {
+ spec => spec.map {
+ case (key, value) => (s"${parentCarbonTable.getTableName}_$key", value)
+ }
+ }
+ CarbonAlterTableDropHivePartitionCommand(
+ tableIdentifier,
+ childSpecs,
+ dropPartitionEvent.ifExists,
+ dropPartitionEvent.purge,
+ dropPartitionEvent.retainData,
+ operationContext)
+ }
+ operationContext.setProperty("dropPartitionCommands", childDropPartitionCommands)
+ childDropPartitionCommands.foreach(_.processMetadata(SparkSession.getActiveSession.get))
+
+ }
+ } else if (parentCarbonTable.isChildDataMap) {
+ if (operationContext.getProperty("isInternalDropCall") == null) {
+ throw new UnsupportedOperationException("Cannot drop partition directly on aggregate table")
+ }
+ }
+ }
+}
+
/**
* below class will be used to create load command for compaction
* for all the pre agregate child data map
@@ -260,9 +472,7 @@ object LoadPostAggregateListener extends OperationEventListener {
* This listener is used to commit all the child data aggregate tables in one transaction. If one
* failes all will be reverted to original state.
*/
-object CommitPreAggregateListener extends OperationEventListener {
-
- private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+object CommitPreAggregateListener extends OperationEventListener with CommitHelper {
override protected def onEvent(event: Event,
operationContext: OperationContext): Unit = {
@@ -317,7 +527,7 @@ object CommitPreAggregateListener extends OperationEventListener {
val backupTableSchemaPath =
CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath) + "_backup_" + uuid
val tableSchemaPath = CarbonTablePath.getTableStatusFilePath(carbonTable.getTablePath)
- markInProgressSegmentAsDeleted(backupTableSchemaPath, operationContext, loadCommand)
+ markInProgressSegmentAsDeleted(backupTableSchemaPath, operationContext, carbonTable)
renameDataMapTableStatusFiles(backupTableSchemaPath, tableSchemaPath, "")
}
}
@@ -333,66 +543,6 @@ object CommitPreAggregateListener extends OperationEventListener {
}
-
- private def markInProgressSegmentAsDeleted(tableStatusFile: String,
- operationContext: OperationContext,
- loadDataCommand: CarbonLoadDataCommand): Unit = {
- val loadMetaDataDetails = SegmentStatusManager.readTableStatusFile(tableStatusFile)
- val segmentBeingLoaded =
- operationContext.getProperty(loadDataCommand.table.getTableUniqueName + "_Segment").toString
- val newDetails = loadMetaDataDetails.collect {
- case detail if detail.getLoadName.equalsIgnoreCase(segmentBeingLoaded) =>
- detail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE)
- detail
- case others => others
- }
- SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusFile, newDetails)
- }
-
- /**
- * Used to rename table status files for commit operation.
- */
- private def renameDataMapTableStatusFiles(sourceFileName: String,
- destinationFileName: String, uuid: String) = {
- val oldCarbonFile = FileFactory.getCarbonFile(sourceFileName)
- val newCarbonFile = FileFactory.getCarbonFile(destinationFileName)
- if (oldCarbonFile.exists() && newCarbonFile.exists()) {
- val backUpPostFix = if (uuid.nonEmpty) {
- "_backup_" + uuid
- } else {
- ""
- }
- LOGGER.info(s"Renaming $newCarbonFile to ${destinationFileName + backUpPostFix}")
- if (newCarbonFile.renameForce(destinationFileName + backUpPostFix)) {
- LOGGER.info(s"Renaming $oldCarbonFile to $destinationFileName")
- oldCarbonFile.renameForce(destinationFileName)
- } else {
- LOGGER.info(s"Renaming $newCarbonFile to ${destinationFileName + backUpPostFix} failed")
- false
- }
- } else {
- false
- }
- }
-
- /**
- * Used to remove table status files with UUID and segment folders.
- */
- private def cleanUpStaleTableStatusFiles(
- childTables: Seq[CarbonTable],
- operationContext: OperationContext,
- uuid: String): Unit = {
- childTables.foreach { childTable =>
- val metaDataDir = FileFactory.getCarbonFile(
- CarbonTablePath.getMetadataPath(childTable.getTablePath))
- val tableStatusFiles = metaDataDir.listFiles(new CarbonFileFilter {
- override def accept(file: CarbonFile): Boolean = {
- file.getName.contains(uuid) || file.getName.contains("backup")
- }
- })
- tableStatusFiles.foreach(_.delete())
- }
- }
}
/**
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateTableHelper.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateTableHelper.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateTableHelper.scala
index 94a8e81..1f9b2b4 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateTableHelper.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateTableHelper.scala
@@ -32,6 +32,8 @@ import org.apache.spark.sql.parser.CarbonSpark2SqlParser
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datamap.Segment
+import org.apache.carbondata.core.metadata.schema.partition.PartitionType
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager}
@@ -62,8 +64,25 @@ case class PreAggregateTableHelper(
val df = sparkSession.sql(updatedQuery)
val fieldRelationMap = PreAggregateUtil.validateActualSelectPlanAndGetAttributes(
df.logicalPlan, queryString)
+ val partitionInfo = parentTable.getPartitionInfo
val fields = fieldRelationMap.keySet.toSeq
val tableProperties = mutable.Map[String, String]()
+ val parentPartitionColumns = if (parentTable.isHivePartitionTable) {
+ partitionInfo.getColumnSchemaList.asScala.map(_.getColumnName)
+ } else {
+ Seq()
+ }
+ // Generate child table partition columns in the same order as the parent table.
+ val partitionerFields = fieldRelationMap.collect {
+ case (field, dataMapField) if parentPartitionColumns
+ .exists(parentCol =>
+ parentCol.equals(dataMapField.columnTableRelationList.get.head.parentColumnName) &&
+ dataMapField.aggregateFunction.isEmpty) =>
+ (PartitionerField(field.name.get,
+ field.dataType,
+ field.columnComment), parentPartitionColumns
+ .indexOf(dataMapField.columnTableRelationList.get.head.parentColumnName))
+ }.toSeq.sortBy(_._2).map(_._1)
dmProperties.foreach(t => tableProperties.put(t._1, t._2))
val selectTable = PreAggregateUtil.getParentCarbonTable(df.logicalPlan)
@@ -77,8 +96,7 @@ case class PreAggregateTableHelper(
fields.filter(col => fieldRelationMap(col).aggregateFunction.isEmpty &&
parentcol.equals(fieldRelationMap(col).
columnTableRelationList.get(0).parentColumnName))
- .map(cols => neworder :+= cols.column)
- )
+ .map(cols => neworder :+= cols.column))
tableProperties.put(CarbonCommonConstants.SORT_COLUMNS, neworder.mkString(","))
tableProperties.put("sort_scope", parentTable.getTableInfo.getFactTable.
getTableProperties.asScala.getOrElse("sort_scope", CarbonCommonConstants
@@ -94,7 +112,7 @@ case class PreAggregateTableHelper(
new CarbonSpark2SqlParser().convertDbNameToLowerCase(tableIdentifier.database),
tableIdentifier.table.toLowerCase,
fields,
- Seq(),
+ partitionerFields,
tableProperties,
None,
isAlterFlow = false,
@@ -185,8 +203,8 @@ case class PreAggregateTableHelper(
val loadAvailable = SegmentStatusManager.readLoadMetadata(parentTable.getMetadataPath)
.collect {
case segment if segment.getSegmentStatus == SegmentStatus.SUCCESS ||
- segment.getSegmentStatus == SegmentStatus.LOAD_PARTIAL_SUCCESS =>
- segment.getLoadName
+ segment.getSegmentStatus == SegmentStatus.LOAD_PARTIAL_SUCCESS =>
+ new Segment(segment.getLoadName, segment.getSegmentFile).toString
}
if (loadAvailable.nonEmpty) {
// Passing segmentToLoad as * because we want to load all the segments into the
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
index f4490ec..04e2135 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
@@ -593,7 +593,6 @@ object PreAggregateUtil {
parentTableIdentifier.table, validateSegments.toString)
CarbonSession.threadSet(CarbonCommonConstants.SUPPORT_DIRECT_QUERY_ON_DATAMAP,
"true")
- CarbonSession.updateSessionInfoToCurrentThread(sparkSession)
try {
loadCommand.processData(sparkSession)
} finally {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
index e700aab..1928b38 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/datasources/SparkCarbonTableFormat.scala
@@ -207,7 +207,9 @@ with Serializable {
taskIdMap.put(path, partitionNumber)
}
val taskID = context.getTaskAttemptID.getTaskID.getId
- CarbonScalaUtil.generateUniqueNumber(taskID, segmentId, partitionNumber)
+ // In case of compaction the segmentId will be like 1.1. Therefore replacing '.' with ""
+ // so that unique number can be generated
+ CarbonScalaUtil.generateUniqueNumber(taskID, segmentId.replace(".", ""), partitionNumber)
}
override def getFileExtension(context: TaskAttemptContext): String = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
index 3344449..e565e68 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/optimizer/CarbonFilters.scala
@@ -443,25 +443,39 @@ object CarbonFilters {
tableIdentifier)
}
+ def getCurrentPartitions(sparkSession: SparkSession,
+ carbonTable: CarbonTable): Option[Seq[PartitionSpec]] = {
+ CarbonFilters.getPartitions(
+ Seq.empty,
+ sparkSession,
+ carbonTable)
+ }
+
+ def getPartitions(partitionFilters: Seq[Expression],
+ sparkSession: SparkSession,
+ identifier: TableIdentifier): Option[Seq[PartitionSpec]] = {
+ val carbonTable = CarbonEnv.getCarbonTable(identifier)(sparkSession)
+ getPartitions(partitionFilters, sparkSession, carbonTable)
+ }
/**
* Fetches partition information from hive
* @param partitionFilters
* @param sparkSession
- * @param identifier
+ * @param carbonTable
* @return
*/
def getPartitions(partitionFilters: Seq[Expression],
sparkSession: SparkSession,
- identifier: TableIdentifier): Option[Seq[PartitionSpec]] = {
- val table = CarbonEnv.getCarbonTable(identifier)(sparkSession)
+ carbonTable: CarbonTable): Option[Seq[PartitionSpec]] = {
+ val identifier = TableIdentifier(carbonTable.getTableName, Some(carbonTable.getDatabaseName))
+ if (!carbonTable.isHivePartitionTable) {
+ return None
+ }
// first try to read partitions in case if the trigger comes from the aggregation table load.
- val partitionsForAggTable = getPartitionsForAggTable(sparkSession, table)
+ val partitionsForAggTable = getPartitionsForAggTable(sparkSession, carbonTable)
if (partitionsForAggTable.isDefined) {
return partitionsForAggTable
}
- if (!table.isHivePartitionTable) {
- return None
- }
val partitions = {
try {
if (CarbonProperties.getInstance().
@@ -503,17 +517,22 @@ object CarbonFilters {
// when validate segments is disabled then only read from partitionmap
val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
if (carbonSessionInfo != null) {
- val validateSegments = carbonSessionInfo.getSessionParams
+ val validateSegments = carbonSessionInfo.getThreadParams
.getProperty(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
table.getDatabaseName + "." +
table.getTableName, "true").toBoolean
if (!validateSegments) {
- val segmentNumbersFromProperty = CarbonProperties.getInstance
+ val segmentNumbersFromProperty = carbonSessionInfo.getThreadParams
.getProperty(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
- table.getDatabaseName + "." + table.getTableName)
- val segment = Segment.toSegment(segmentNumbersFromProperty)
- val segmentFile = new SegmentFileStore(table.getTablePath, segment.getSegmentFileName)
- Some(segmentFile.getPartitionSpecs.asScala)
+ table.getDatabaseName + "." + table.getTableName, "*")
+ // In case of compaction multiple segments will be passed as CARBON_INPUT_SEGMENTS.
+ // Therefore partitionSpec will be extracted from all segments.
+ val segments = segmentNumbersFromProperty.split(",").flatMap { a =>
+ val segment = Segment.toSegment(a)
+ val segmentFile = new SegmentFileStore(table.getTablePath, segment.getSegmentFileName)
+ segmentFile.getPartitionSpecs.asScala
+ }
+ Some(segments.toSet.toSeq)
} else {
None
}
[2/2] carbondata git commit: [CARBONDATA-2294] Partition preaggregate
support
Posted by ma...@apache.org.
[CARBONDATA-2294] Partition preaggregate support
1. Allow creation of aggregate table on partition and non-partition columns
2. Support to load data into aggregate table after insertion into parent table is complete.
3. Support to compact aggregate tables if parent table is compacted.
4. If parent table partition is dropped then drop the same partition in child tables also.
This closes #2109
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/55084602
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/55084602
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/55084602
Branch: refs/heads/master
Commit: 55084602917aec5a2ff5c29e6e7ac501d7830d14
Parents: 248e0c8
Author: kunal642 <ku...@gmail.com>
Authored: Fri Mar 23 12:12:05 2018 +0530
Committer: manishgupta88 <to...@gmail.com>
Committed: Thu Apr 5 15:31:39 2018 +0530
----------------------------------------------------------------------
.../core/metadata/SegmentFileStore.java | 6 +-
.../core/mutate/CarbonUpdateUtil.java | 9 +-
.../carbondata/core/util/CarbonSessionInfo.java | 18 +-
.../carbondata/core/util/SessionParams.java | 10 +-
.../hadoop/api/CarbonOutputCommitter.java | 25 +-
.../TestPreAggregateCompaction.scala | 15 +
...ndardPartitionWithPreaggregateTestCase.scala | 503 +++++++++++++++++++
.../carbondata/events/AlterTableEvents.scala | 14 +-
.../org/apache/carbondata/events/Events.scala | 10 +
.../carbondata/spark/rdd/CarbonScanRDD.scala | 44 +-
.../carbondata/spark/rdd/StreamHandoffRDD.scala | 8 +-
.../org/apache/spark/util/PartitionUtils.scala | 2 +-
.../spark/rdd/AggregateDataMapCompactor.scala | 14 +-
.../spark/rdd/CarbonDataRDDFactory.scala | 2 +-
.../spark/rdd/CarbonTableCompactor.scala | 4 +-
.../scala/org/apache/spark/sql/CarbonEnv.scala | 5 +
.../management/CarbonLoadDataCommand.scala | 35 +-
...arbonAlterTableAddHivePartitionCommand.scala | 8 +-
...rbonAlterTableDropHivePartitionCommand.scala | 51 +-
.../preaaggregate/PreAggregateListeners.scala | 280 ++++++++---
.../preaaggregate/PreAggregateTableHelper.scala | 28 +-
.../preaaggregate/PreAggregateUtil.scala | 1 -
.../datasources/SparkCarbonTableFormat.scala | 4 +-
.../spark/sql/optimizer/CarbonFilters.scala | 45 +-
24 files changed, 966 insertions(+), 175 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java b/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
index 257ee4c..5902148 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/SegmentFileStore.java
@@ -579,13 +579,15 @@ public class SegmentFileStore {
* @throws IOException
*/
public static void commitDropPartitions(CarbonTable carbonTable, String uniqueId,
- List<String> toBeUpdatedSegments, List<String> toBeDeleteSegments) throws IOException {
+ List<String> toBeUpdatedSegments, List<String> toBeDeleteSegments,
+ String uuid) throws IOException {
if (toBeDeleteSegments.size() > 0 || toBeUpdatedSegments.size() > 0) {
Set<Segment> segmentSet = new HashSet<>(
new SegmentStatusManager(carbonTable.getAbsoluteTableIdentifier())
.getValidAndInvalidSegments().getValidSegments());
CarbonUpdateUtil.updateTableMetadataStatus(segmentSet, carbonTable, uniqueId, true,
- Segment.toSegmentList(toBeDeleteSegments), Segment.toSegmentList(toBeUpdatedSegments));
+ Segment.toSegmentList(toBeDeleteSegments), Segment.toSegmentList(toBeUpdatedSegments),
+ uuid);
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
index 2b4dabe..93e5580 100644
--- a/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
+++ b/core/src/main/java/org/apache/carbondata/core/mutate/CarbonUpdateUtil.java
@@ -184,7 +184,7 @@ public class CarbonUpdateUtil {
CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdationRequired,
List<Segment> segmentsToBeDeleted) {
return updateTableMetadataStatus(updatedSegmentsList, table, updatedTimeStamp,
- isTimestampUpdationRequired, segmentsToBeDeleted, new ArrayList<Segment>());
+ isTimestampUpdationRequired, segmentsToBeDeleted, new ArrayList<Segment>(), "");
}
/**
@@ -198,12 +198,13 @@ public class CarbonUpdateUtil {
*/
public static boolean updateTableMetadataStatus(Set<Segment> updatedSegmentsList,
CarbonTable table, String updatedTimeStamp, boolean isTimestampUpdationRequired,
- List<Segment> segmentsToBeDeleted, List<Segment> segmentFilesTobeUpdated) {
+ List<Segment> segmentsToBeDeleted, List<Segment> segmentFilesTobeUpdated, String uuid) {
boolean status = false;
String metaDataFilepath = table.getMetadataPath();
AbsoluteTableIdentifier identifier = table.getAbsoluteTableIdentifier();
- String tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath());
+ String tableStatusPath =
+ CarbonTablePath.getTableStatusFilePathWithUUID(identifier.getTablePath(), uuid);
SegmentStatusManager segmentStatusManager = new SegmentStatusManager(identifier);
ICarbonLock carbonLock = segmentStatusManager.getTableStatusLock();
@@ -570,7 +571,7 @@ public class CarbonUpdateUtil {
UUID,
false,
new ArrayList<Segment>(),
- segmentFilesToBeUpdatedLatest);
+ segmentFilesToBeUpdatedLatest, "");
}
// delete the update table status files which are old.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java b/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java
index e397e55..25d1c29 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/CarbonSessionInfo.java
@@ -21,8 +21,6 @@ import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
-import org.apache.carbondata.core.exception.InvalidConfigurationException;
-
/**
* This class maintains carbon session information details
*/
@@ -60,20 +58,8 @@ public class CarbonSessionInfo implements Serializable, Cloneable {
public CarbonSessionInfo clone() throws CloneNotSupportedException {
super.clone();
CarbonSessionInfo newObj = new CarbonSessionInfo();
- for (Map.Entry<String, String> entry : sessionParams.getAll().entrySet()) {
- try {
- newObj.getSessionParams().addProperty(entry.getKey(), entry.getValue(), false);
- } catch (InvalidConfigurationException ex) {
- ex.printStackTrace();
- }
- }
- for (Map.Entry<String, String> entry : threadParams.getAll().entrySet()) {
- try {
- newObj.getThreadParams().addProperty(entry.getKey(), entry.getValue(), false);
- } catch (InvalidConfigurationException ex) {
- ex.printStackTrace();
- }
- }
+ newObj.setSessionParams(sessionParams.clone());
+ newObj.setThreadParams(threadParams.clone());
Map<String, Object> nonSerializableExtraInfo = getNonSerializableExtraInfo();
for (Map.Entry<String, Object> entry : nonSerializableExtraInfo.entrySet()) {
nonSerializableExtraInfo.put(entry.getKey(), entry.getValue());
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java b/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
index 20968a3..63af23a 100644
--- a/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
+++ b/core/src/main/java/org/apache/carbondata/core/util/SessionParams.java
@@ -47,7 +47,7 @@ import static org.apache.carbondata.core.constants.CarbonLoadOptionConstants.CAR
/**
* This class maintains carbon session params
*/
-public class SessionParams implements Serializable {
+public class SessionParams implements Serializable, Cloneable {
private static final LogService LOGGER =
LogServiceFactory.getLogService(CacheProvider.class.getName());
@@ -228,4 +228,12 @@ public class SessionParams implements Serializable {
sProps.clear();
}
+ public SessionParams clone() throws CloneNotSupportedException {
+ super.clone();
+ SessionParams newObj = new SessionParams();
+ newObj.addedProps.putAll(this.addedProps);
+ newObj.sProps.putAll(this.sProps);
+ newObj.extraInfo.putAll(this.extraInfo);
+ return newObj;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
index c6d1f0b..0add279 100644
--- a/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
+++ b/hadoop/src/main/java/org/apache/carbondata/hadoop/api/CarbonOutputCommitter.java
@@ -130,6 +130,12 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
loadModel.getTablePath());
newMetaEntry.setSegmentFile(segmentFileName + CarbonTablePath.SEGMENT_EXT);
}
+ OperationContext operationContext = (OperationContext) getOperationContext();
+ String uuid = "";
+ if (loadModel.getCarbonDataLoadSchema().getCarbonTable().isChildDataMap() &&
+ operationContext != null) {
+ uuid = operationContext.getProperty("uuid").toString();
+ }
CarbonLoaderUtil
.populateNewLoadMetaEntry(newMetaEntry, SegmentStatus.SUCCESS, loadModel.getFactTimeStamp(),
true);
@@ -137,15 +143,14 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
long segmentSize = CarbonLoaderUtil
.addDataIndexSizeIntoMetaEntry(newMetaEntry, loadModel.getSegmentId(), carbonTable);
if (segmentSize > 0 || overwriteSet) {
- Object operationContext = getOperationContext();
- if (operationContext != null) {
- ((OperationContext) operationContext)
+ if (operationContext != null && carbonTable.hasAggregationDataMap()) {
+ operationContext
.setProperty("current.segmentfile", newMetaEntry.getSegmentFile());
LoadEvents.LoadTablePreStatusUpdateEvent event =
new LoadEvents.LoadTablePreStatusUpdateEvent(carbonTable.getCarbonTableIdentifier(),
loadModel);
try {
- OperationListenerBus.getInstance().fireEvent(event, (OperationContext) operationContext);
+ OperationListenerBus.getInstance().fireEvent(event, operationContext);
} catch (Exception e) {
throw new IOException(e);
}
@@ -155,9 +160,9 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
if (segmentSize == 0) {
newMetaEntry.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE);
}
- uniqueId = overwritePartitions(loadModel, newMetaEntry);
+ uniqueId = overwritePartitions(loadModel, newMetaEntry, uuid);
} else {
- CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false);
+ CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid);
}
DataMapStatusManager.disableDataMapsOfTable(carbonTable);
if (operationContext != null) {
@@ -165,7 +170,7 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
new LoadEvents.LoadTablePostStatusUpdateEvent(loadModel);
try {
OperationListenerBus.getInstance()
- .fireEvent(postStatusUpdateEvent, (OperationContext) operationContext);
+ .fireEvent(postStatusUpdateEvent, operationContext);
} catch (Exception e) {
throw new IOException(e);
}
@@ -201,8 +206,8 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
* @return
* @throws IOException
*/
- private String overwritePartitions(CarbonLoadModel loadModel, LoadMetadataDetails newMetaEntry)
- throws IOException {
+ private String overwritePartitions(CarbonLoadModel loadModel, LoadMetadataDetails newMetaEntry,
+ String uuid) throws IOException {
CarbonTable table = loadModel.getCarbonDataLoadSchema().getCarbonTable();
SegmentFileStore fileStore = new SegmentFileStore(loadModel.getTablePath(),
loadModel.getSegmentId() + "_" + loadModel.getFactTimeStamp()
@@ -224,7 +229,7 @@ public class CarbonOutputCommitter extends FileOutputCommitter {
}
newMetaEntry.setUpdateStatusFileName(uniqueId);
// Commit the removed partitions in carbon store.
- CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, "",
+ CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, loadModel, false, false, uuid,
Segment.toSegmentList(tobeDeletedSegs), Segment.toSegmentList(tobeUpdatedSegs));
return uniqueId;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateCompaction.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateCompaction.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateCompaction.scala
index 42209b3..d794152 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateCompaction.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateCompaction.scala
@@ -21,6 +21,9 @@ import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.scalatest.Matchers._
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.CarbonProperties
+
class TestPreAggregateCompaction extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll {
val testData = s"$resourcesPath/sample.csv"
@@ -176,7 +179,19 @@ class TestPreAggregateCompaction extends QueryTest with BeforeAndAfterEach with
segmentNamesSum.sorted should equal (Array("0", "0.1", "0.2", "1", "2", "3", "4", "5", "6", "7"))
}
+ test("test auto compaction on aggregate table") {
+ CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "true")
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
+ val segmentNamesSum = sql("show segments for table maintable_preagg_sum").collect().map(_.get(0).toString)
+ segmentNamesSum.sorted should equal (Array("0", "0.1", "1", "2", "3"))
+ CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "false")
+ }
+
override def afterAll(): Unit = {
+ CarbonProperties.getInstance().addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "false")
sql("drop database if exists compaction cascade")
sql("use default")
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionWithPreaggregateTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionWithPreaggregateTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionWithPreaggregateTestCase.scala
new file mode 100644
index 0000000..02c3158
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/standardpartition/StandardPartitionWithPreaggregateTestCase.scala
@@ -0,0 +1,503 @@
+/*
+* 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 id 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.standardpartition
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.datasources.LogicalRelation
+import org.apache.spark.sql.hive.CarbonRelation
+import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, Row}
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.BeforeAndAfterAll
+
+class StandardPartitionWithPreaggregateTestCase extends QueryTest with BeforeAndAfterAll {
+
+ val testData = s"$resourcesPath/sample.csv"
+
+ override def beforeAll(): Unit = {
+ sql("drop database if exists partition_preaggregate cascade")
+ sql("create database partition_preaggregate")
+ sql("use partition_preaggregate")
+ sql(
+ """
+ | CREATE TABLE par(id INT, name STRING, age INT) PARTITIONED BY(city STRING)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ """
+ | CREATE TABLE maintable(id int, name string, city string) partitioned by (age int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(s"LOAD DATA LOCAL INPATH '$testData' into table maintable")
+ }
+
+ override def afterAll(): Unit = {
+ sql("drop database if exists partition_preaggregate cascade")
+ sql("use default")
+ }
+
+ // Create aggregate table on partition with partition column in aggregation only.
+ test("test preaggregate table creation on partition table with partition col as aggregation") {
+ sql("create datamap p1 on table par using 'preaggregate' as select id, sum(city) from par group by id")
+ assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p1")(sqlContext.sparkSession).isHivePartitionTable)
+ }
+
+ // Create aggregate table on partition with partition column in projection and aggregation only.
+ test("test preaggregate table creation on partition table with partition col as projection") {
+ sql("create datamap p2 on table par using 'preaggregate' as select id, city, min(city) from par group by id,city ")
+ assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p2")(sqlContext.sparkSession).isHivePartitionTable)
+ }
+
+ // Create aggregate table on partition with partition column as group by.
+ test("test preaggregate table creation on partition table with partition col as group by") {
+ sql("create datamap p3 on table par using 'preaggregate' as select id, max(city) from par group by id,city ")
+ assert(CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p3")(sqlContext.sparkSession).isHivePartitionTable)
+ }
+
+ // Create aggregate table on partition without partition column.
+ test("test preaggregate table creation on partition table without partition column") {
+ sql("create datamap p4 on table par using 'preaggregate' as select name, count(id) from par group by name ")
+ assert(!CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "par_p4")(sqlContext.sparkSession).isHivePartitionTable)
+ }
+
+ test("test data correction in aggregate table when partition column is used") {
+ sql("create datamap p1 on table maintable using 'preaggregate' as select id, sum(age) from maintable group by id, age")
+ checkAnswer(sql("select * from maintable_p1"),
+ Seq(Row(1,31,31),
+ Row(2,27,27),
+ Row(3,70,35),
+ Row(4,26,26),
+ Row(4,29,29)))
+ preAggTableValidator(sql("select id, sum(age) from maintable group by id, age").queryExecution.analyzed, "maintable_p1")
+ sql("drop datamap p1 on table maintable")
+ }
+
+ test("test data correction in aggregate table when partition column is not used") {
+ sql("create datamap p2 on table maintable using 'preaggregate' as select id, max(age) from maintable group by id")
+ checkAnswer(sql("select * from maintable_p2"),
+ Seq(Row(1,31),
+ Row(2,27),
+ Row(3,35),
+ Row(4,29)))
+ preAggTableValidator(sql("select id, max(age) from maintable group by id").queryExecution.analyzed, "maintable_p2")
+ sql("drop datamap p2 on table maintable")
+ }
+
+ test("test data correction with insert overwrite") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert overwrite table partitionone values('v',2014,1,1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("v",2014,1,1)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("v",2014,2014,1,1)))
+ }
+
+ test("test data correction with insert overwrite on different value") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert overwrite table partitionone values('v',2015,1,1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,1), Row("v",2015,1,1)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,1), Row("v",2015,2015,1,1)))
+ }
+
+ test("test to check column ordering in parent and child table") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month, year,day")
+ val parentTable = CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "partitionone")(sqlContext.sparkSession)
+ val childTable = CarbonEnv.getCarbonTable(Some("partition_preaggregate"), "partitionone_p1")(sqlContext.sparkSession)
+ val parentPartitionColumns = parentTable.getPartitionInfo.getColumnSchemaList
+ val childPartitionColumns = childTable.getPartitionInfo.getColumnSchemaList
+ assert(parentPartitionColumns.asScala.zip(childPartitionColumns.asScala).forall {
+ case (a,b) =>
+ a.getColumnName.equalsIgnoreCase(b.getParentColumnTableRelations.get(0).getColumnName)
+ })
+ }
+
+ test("test data after minor compaction on partition table with pre-aggregate") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone compact 'minor'")
+ val showSegments = sql("show segments for table partitionone").collect().map{a=> (a.get(0), a.get(1))}
+ assert(showSegments.count(_._2 == "Success") == 1)
+ assert(showSegments.count(_._2 == "Compacted") == 4)
+ }
+
+ test("test data after major compaction on partition table with pre-aggregate") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month,day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone compact 'major'")
+ val showSegments = sql("show segments for table partitionone").collect().map{a=> (a.get(0), a.get(1))}
+ assert(showSegments.count(_._2 == "Success") == 1)
+ assert(showSegments.count(_._2 == "Compacted") == 5)
+ }
+
+ test("test drop partition 1") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(day=1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2)))
+ }
+
+ test("test drop partition 2") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,3)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(day=1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015,2,3)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3)))
+ }
+
+ test("test data with filter query") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,3)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(day=1)")
+ checkAnswer(sql("select empname, sum(year) from partitionone where day=3 group by empname, year, month, day"), Seq(Row("k",2015)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3)))
+ }
+
+ test("test drop partition 3") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,3)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(day=1,month=1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3), Row("k",2015,2015,2,1)))
+ }
+
+ test("test drop partition 4") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,3)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(year=2014,day=1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015, 2,3), Row("k",2015,2015, 2,1)))
+ }
+
+ test("test drop partition 5") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,3)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3), Row("k",2015,2015,2,1)))
+ sql("show partitions partitionone_p1").show()
+ }
+
+ test("test drop partition 6") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql("create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2015,2,3)")
+ sql("insert into partitionone values('k',2015,2,1)")
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+ checkAnswer(sql("select * from partitionone"), Seq(Row("k",2014,1,2), Row("k",2015, 2,3), Row("k",2015, 2,1)))
+ checkAnswer(sql("select * from partitionone_p1"), Seq(Row("k",2014,2014,1,2), Row("k",2015,2015,2,3), Row("k",2015,2015,2,1)))
+ }
+
+ test("test drop partition 7") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, day")
+ sql(
+ "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ val exceptionMessage = intercept[Exception] {
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+ }.getMessage
+ assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
+ assert(exceptionMessage.contains("p2"))
+ assert(exceptionMessage.contains("p1"))
+ }
+
+ test("test drop partition 8") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql(
+ "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, month, day")
+ sql(
+ "create datamap p3 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ val exceptionMessage = intercept[Exception] {
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+ }.getMessage
+ assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
+ assert(exceptionMessage.contains("p2"))
+ assert(exceptionMessage.contains("p3"))
+ assert(!exceptionMessage.contains("p1"))
+ }
+
+ test("test drop partition 9") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ val exceptionMessage = intercept[Exception] {
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+ }.getMessage
+ assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
+ assert(exceptionMessage.contains("p1"))
+ }
+
+ test("test drop partition 10") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
+ sql(
+ "create datamap p2 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year, month, day")
+ sql("insert into partitionone values('k',2014,1,1)")
+ sql("insert into partitionone values('k',2014,1,2)")
+ val exceptionMessage = intercept[Exception] {
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+ }.getMessage
+ assert(exceptionMessage.contains("Cannot drop partition as one of the partition"))
+ assert(exceptionMessage.contains("p1"))
+ sql("drop datamap p1 on table partitionone")
+ sql("alter table partitionone drop partition(year=2014,month=1, day=1)")
+ }
+
+ test("test drop partition 11") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname, year")
+ sql("insert into partitionone values('k',2014,1,1)")
+ val exceptionMessage = intercept[Exception] {
+ sql("alter table partitionone_p1 drop partition(partitionone_year=1)")
+ }.getMessage
+ assert(exceptionMessage.contains("Cannot drop partition directly on aggregate table"))
+ }
+
+ test("test drop partition 12") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
+ sql("insert into partitionone values('k',2014,1,1)")
+ val exceptionMessage = intercept[Exception] {
+ sql("alter table partitionone_p1 drop partition(year=2014)")
+ }.getMessage
+ assert(exceptionMessage.contains("operation failed for partition_preaggregate.partitionone_p1: Not a partitioned table"))
+ }
+
+ test("test add partition on aggregate table") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
+ assert(intercept[Exception] {
+ sql("alter table partitionone_p1 add partition(c=1)")
+ }.getMessage.equals("Cannot add partition directly on non partitioned table"))
+ }
+
+ test("test if alter rename is blocked on partition table with preaggregate") {
+ sql("drop table if exists partitionone")
+ sql(
+ """
+ | CREATE TABLE if not exists partitionone (empname String, id int)
+ | PARTITIONED BY (year int, month int,day int)
+ | STORED BY 'org.apache.carbondata.format'
+ """.stripMargin)
+ sql(
+ "create datamap p1 on table partitionone using 'preaggregate' as select empname, sum(year) from partitionone group by empname")
+ intercept[Exception] {
+ sql("alter table partitionone rename to p")
+ }
+ }
+
+ def preAggTableValidator(plan: LogicalPlan, actualTableName: String) : Unit = {
+ var isValidPlan = false
+ plan.transform {
+ // first check if any preaTable1 scala function is applied it is present is in plan
+ // then call is from create preaTable1regate table class so no need to transform the query plan
+ case ca:CarbonRelation =>
+ if (ca.isInstanceOf[CarbonDatasourceHadoopRelation]) {
+ val relation = ca.asInstanceOf[CarbonDatasourceHadoopRelation]
+ if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
+ isValidPlan = true
+ }
+ }
+ ca
+ case logicalRelation:LogicalRelation =>
+ if(logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation]) {
+ val relation = logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
+ if(relation.carbonTable.getTableName.equalsIgnoreCase(actualTableName)) {
+ isValidPlan = true
+ }
+ }
+ logicalRelation
+ }
+ assert(isValidPlan)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
index 7c4339f..dbd9ebf 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
@@ -17,8 +17,10 @@
package org.apache.carbondata.events
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.execution.command._
+import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
@@ -81,7 +83,6 @@ case class AlterTableDropColumnAbortEvent(
alterTableDropColumnModel: AlterTableDropColumnModel,
sparkSession: SparkSession) extends Event with AlterTableDropColumnEventInfo
-
/**
*
* @param carbonTable
@@ -231,3 +232,14 @@ case class PreAlterTableHivePartitionCommandEvent(sparkSession: SparkSession,
*/
case class PostAlterTableHivePartitionCommandEvent(sparkSession: SparkSession,
carbonTable: CarbonTable) extends Event with AlterTableHivePartitionInfo
+
+case class AlterTableDropPartitionMetaEvent(parentCarbonTable: CarbonTable,
+ specs: Seq[TablePartitionSpec],
+ ifExists: Boolean,
+ purge: Boolean,
+ retainData: Boolean)
+ extends Event with AlterTableDropPartitionEventInfo
+
+case class AlterTableDropPartitionPreStatusEvent(carbonTable: CarbonTable) extends Event
+
+case class AlterTableDropPartitionPostStatusEvent(carbonTable: CarbonTable) extends Event
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
index d85e8ae..da62e02 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
@@ -18,8 +18,10 @@
package org.apache.carbondata.events
import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
import org.apache.spark.sql.execution.command.{AlterTableAddColumnsModel, AlterTableDataTypeChangeModel, AlterTableDropColumnModel, AlterTableRenameModel, CarbonMergerMapping}
+import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
@@ -69,6 +71,14 @@ trait AlterTableDropColumnEventInfo {
val alterTableDropColumnModel: AlterTableDropColumnModel
}
+trait AlterTableDropPartitionEventInfo {
+ val parentCarbonTable: CarbonTable
+ val specs: Seq[TablePartitionSpec]
+ val ifExists: Boolean
+ val purge: Boolean
+ val retainData: Boolean
+}
+
trait AlterTableDataTypeChangeEventInfo {
val carbonTable: CarbonTable
val alterTableDataTypeChangeModel: AlterTableDataTypeChangeModel
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
index efb20eb..8c433ea 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/CarbonScanRDD.scala
@@ -41,6 +41,7 @@ import org.apache.spark.sql.util.SparkSQLUtil.sessionState
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonCommonConstantsInternal}
+import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.datastore.block.Distributable
import org.apache.carbondata.core.indexstore.PartitionSpec
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
@@ -52,7 +53,7 @@ import org.apache.carbondata.core.stats.{QueryStatistic, QueryStatisticsConstant
import org.apache.carbondata.core.statusmanager.FileFormat
import org.apache.carbondata.core.util._
import org.apache.carbondata.hadoop._
-import org.apache.carbondata.hadoop.api.{CarbonFileInputFormat, CarbonInputFormat, CarbonTableInputFormat}
+import org.apache.carbondata.hadoop.api.{CarbonFileInputFormat, CarbonInputFormat}
import org.apache.carbondata.hadoop.api.CarbonTableInputFormat
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.spark.InitInputMetrics
@@ -534,10 +535,29 @@ class CarbonScanRDD(
// when validate segments is disabled in thread local update it to CarbonTableInputFormat
val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
if (carbonSessionInfo != null) {
- CarbonInputFormat.setValidateSegmentsToAccess(conf, carbonSessionInfo.getSessionParams
- .getProperty(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
- identifier.getCarbonTableIdentifier.getDatabaseName + "." +
- identifier.getCarbonTableIdentifier.getTableName, "true").toBoolean)
+ val tableUniqueKey = identifier.getDatabaseName + "." + identifier.getTableName
+ val validateInputSegmentsKey = CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
+ tableUniqueKey
+ CarbonInputFormat.setValidateSegmentsToAccess(conf, carbonSessionInfo.getThreadParams
+ .getProperty(validateInputSegmentsKey, "true").toBoolean)
+ val queryOnPreAggStreamingKey = CarbonCommonConstantsInternal.QUERY_ON_PRE_AGG_STREAMING +
+ tableUniqueKey
+ val queryOnPreAggStreaming = carbonSessionInfo.getThreadParams
+ .getProperty(queryOnPreAggStreamingKey, "false").toBoolean
+ val inputSegmentsKey = CarbonCommonConstants.CARBON_INPUT_SEGMENTS + tableUniqueKey
+ CarbonInputFormat.setValidateSegmentsToAccess(conf, carbonSessionInfo.getThreadParams
+ .getProperty(validateInputSegmentsKey, "true").toBoolean)
+ CarbonInputFormat
+ .setQuerySegment(conf,
+ carbonSessionInfo.getThreadParams
+ .getProperty(inputSegmentsKey,
+ CarbonProperties.getInstance().getProperty(inputSegmentsKey, "*")))
+ if (queryOnPreAggStreaming) {
+ CarbonInputFormat.setAccessStreamingSegments(conf, queryOnPreAggStreaming)
+ carbonSessionInfo.getThreadParams.removeProperty(queryOnPreAggStreamingKey)
+ carbonSessionInfo.getThreadParams.removeProperty(inputSegmentsKey)
+ carbonSessionInfo.getThreadParams.removeProperty(validateInputSegmentsKey)
+ }
}
format
}
@@ -562,20 +582,22 @@ class CarbonScanRDD(
val tableUniqueKey = identifier.getDatabaseName + "." + identifier.getTableName
val validateInputSegmentsKey = CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
tableUniqueKey
- CarbonInputFormat.setValidateSegmentsToAccess(conf, carbonSessionInfo.getSessionParams
- .getProperty(validateInputSegmentsKey, "true").toBoolean)
+ CarbonInputFormat.setValidateSegmentsToAccess(conf, carbonSessionInfo.getThreadParams
+ .getProperty(validateInputSegmentsKey, "true").toBoolean)
val queryOnPreAggStreamingKey = CarbonCommonConstantsInternal.QUERY_ON_PRE_AGG_STREAMING +
- tableUniqueKey
+ tableUniqueKey
val queryOnPreAggStreaming = carbonSessionInfo.getThreadParams
.getProperty(queryOnPreAggStreamingKey, "false").toBoolean
- CarbonInputFormat.setAccessStreamingSegments(conf, queryOnPreAggStreaming)
val inputSegmentsKey = CarbonCommonConstants.CARBON_INPUT_SEGMENTS + tableUniqueKey
CarbonInputFormat.setValidateSegmentsToAccess(conf, carbonSessionInfo.getThreadParams
.getProperty(validateInputSegmentsKey, "true").toBoolean)
CarbonInputFormat
.setQuerySegment(conf,
- carbonSessionInfo.getThreadParams.getProperty(inputSegmentsKey, "*"))
- if(queryOnPreAggStreaming) {
+ carbonSessionInfo.getThreadParams
+ .getProperty(inputSegmentsKey,
+ CarbonProperties.getInstance().getProperty(inputSegmentsKey, "*")))
+ if (queryOnPreAggStreaming) {
+ CarbonInputFormat.setAccessStreamingSegments(conf, queryOnPreAggStreaming)
carbonSessionInfo.getThreadParams.removeProperty(queryOnPreAggStreamingKey)
carbonSessionInfo.getThreadParams.removeProperty(inputSegmentsKey)
carbonSessionInfo.getThreadParams.removeProperty(validateInputSegmentsKey)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
index f57d70c..3cf9c55 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/rdd/StreamHandoffRDD.scala
@@ -37,7 +37,7 @@ import org.apache.carbondata.core.metadata.CarbonMetadata
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.scan.result.iterator.RawResultIterator
import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
-import org.apache.carbondata.core.util.CarbonUtil
+import org.apache.carbondata.core.util.{CarbonUtil, DataTypeUtil}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonProjection}
@@ -47,7 +47,7 @@ import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.processing.merger.{CompactionResultSortProcessor, CompactionType}
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.spark.{HandoffResult, HandoffResultImpl}
-import org.apache.carbondata.spark.util.CommonUtil
+import org.apache.carbondata.spark.util.{CommonUtil, SparkDataTypeConverterImpl}
import org.apache.carbondata.streaming.{CarbonStreamInputFormat, CarbonStreamRecordReader}
@@ -111,6 +111,7 @@ class StreamHandoffRDD[K, V](
): Iterator[(K, V)] = {
carbonLoadModel.setTaskNo("" + split.index)
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+ DataTypeUtil.setDataTypeConverter(new SparkDataTypeConverterImpl)
CarbonMetadata.getInstance().addCarbonTable(carbonTable)
// the input iterator is using raw row
val iteratorList = prepareInputIterator(split, carbonTable)
@@ -316,6 +317,7 @@ object StreamHandoffRDD {
false)
CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, carbonLoadModel, true, false)
// convert a streaming segment to columnar segment
+
val status = new StreamHandoffRDD(
sparkSession.sparkContext,
new HandoffResultImpl(),
@@ -353,7 +355,9 @@ object StreamHandoffRDD {
carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getCarbonTableIdentifier,
carbonLoadModel)
OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)
+
val done = updateLoadMetadata(handoffSegmenId, carbonLoadModel)
+
val loadTablePostStatusUpdateEvent: LoadTablePostStatusUpdateEvent =
new LoadTablePostStatusUpdateEvent(carbonLoadModel)
OperationListenerBus.getInstance()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/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 37cd12e..605777d 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
@@ -218,7 +218,7 @@ object PartitionUtils {
alterPartitionModel.carbonLoadModel.getFactTimeStamp.toString,
true,
new util.ArrayList[Segment](0),
- new util.ArrayList[Segment](segmentFiles))) {
+ new util.ArrayList[Segment](segmentFiles), "")) {
throw new IOException("Data update failed due to failure in table status updation.")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/AggregateDataMapCompactor.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/AggregateDataMapCompactor.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/AggregateDataMapCompactor.scala
index 4266c51..dd2b097 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/AggregateDataMapCompactor.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/AggregateDataMapCompactor.scala
@@ -26,6 +26,7 @@ import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.execution.command.preaaggregate.PreAggregateUtil
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager}
import org.apache.carbondata.core.util.path.CarbonTablePath
@@ -47,7 +48,13 @@ class AggregateDataMapCompactor(carbonLoadModel: CarbonLoadModel,
override def executeCompaction(): Unit = {
val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
val loadMetaDataDetails = identifySegmentsToBeMerged()
- val segments = loadMetaDataDetails.asScala.map(_.getLoadName)
+ // If segmentFile name is specified in load details then segment is for partition table
+ // therefore the segment file name should be loadName#segmentFileName.segment
+ val segments = loadMetaDataDetails.asScala.map {
+ loadDetail =>
+ new Segment(loadDetail.getLoadName, loadDetail.getSegmentFile).toString
+ }
+
if (segments.nonEmpty) {
val mergedLoadName = CarbonDataMergerUtil.getMergedLoadName(loadMetaDataDetails).split("_")(1)
CarbonSession.threadSet(
@@ -59,7 +66,6 @@ class AggregateDataMapCompactor(carbonLoadModel: CarbonLoadModel,
CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
carbonLoadModel.getDatabaseName + "." +
carbonLoadModel.getTableName, "false")
- CarbonSession.updateSessionInfoToCurrentThread(sqlContext.sparkSession)
val loadCommand = operationContext.getProperty(carbonTable.getTableName + "_Compaction")
.asInstanceOf[CarbonLoadDataCommand]
val uuid = Option(loadCommand.operationContext.getProperty("uuid")).getOrElse("").toString
@@ -121,6 +127,10 @@ class AggregateDataMapCompactor(carbonLoadModel: CarbonLoadModel,
CarbonSession.threadUnset(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS +
carbonLoadModel.getDatabaseName + "." +
carbonLoadModel.getTableName)
+ LOGGER
+ .info(s"Compaction request for datamap ${ carbonTable.getTableUniqueName } is successful")
+ LOGGER
+ .audit(s"Compaction request for datamap ${carbonTable.getTableUniqueName} is successful")
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/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 a4b30da..65e5e51 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
@@ -449,7 +449,7 @@ object CarbonDataRDDFactory {
updateModel.get.updatedTimeStamp + "",
true,
new util.ArrayList[Segment](0),
- new util.ArrayList[Segment](segmentFiles))) {
+ new util.ArrayList[Segment](segmentFiles), "")) {
LOGGER.audit("Data update is successful for " +
s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
} else {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
index 15ae30f..91e2335 100644
--- a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
+++ b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonTableCompactor.scala
@@ -238,6 +238,8 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
carbonLoadModel.getFactTimeStamp.toString)
}
}
+ // Used to inform the commit listener that the commit is fired from compaction flow.
+ operationContext.setProperty("isCompaction", "true")
// trigger event for compaction
val alterTableCompactionPreStatusUpdateEvent =
AlterTableCompactionPreStatusUpdateEvent(sc.sparkSession,
@@ -270,8 +272,6 @@ class CarbonTableCompactor(carbonLoadModel: CarbonLoadModel,
carbonMergerMapping,
carbonLoadModel,
mergedLoadName)
- // Used to inform the commit listener that the commit is fired from compaction flow.
- operationContext.setProperty("isCompaction", "true")
val commitComplete = try {
// Once main table compaction is done and 0.1, 4.1, 8.1 is created commit will happen for
// all the tables. The commit listener will compact the child tables until no more segments
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
index 63e0834..881a4f5 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonEnv.scala
@@ -170,6 +170,11 @@ object CarbonEnv {
.addListener(classOf[LoadMetadataEvent], CompactionProcessMetaListener)
.addListener(classOf[LoadTablePostStatusUpdateEvent], CommitPreAggregateListener)
.addListener(classOf[AlterTableCompactionPostStatusUpdateEvent], CommitPreAggregateListener)
+ .addListener(classOf[AlterTableDropPartitionPreStatusEvent],
+ AlterTableDropPartitionPreStatusListener)
+ .addListener(classOf[AlterTableDropPartitionPostStatusEvent],
+ AlterTableDropPartitionPostStatusListener)
+ .addListener(classOf[AlterTableDropPartitionMetaEvent], AlterTableDropPartitionMetaListener)
}
def registerCommonListener(sparkSession: SparkSession): Unit = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
index e0736fb..f0a9a9e 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -130,12 +130,6 @@ case class CarbonLoadDataCommand(
case l: LogicalRelation => l
}.head
sizeInBytes = logicalPartitionRelation.relation.sizeInBytes
- currPartitions = CarbonFilters.getCurrentPartitions(
- sparkSession,
- TableIdentifier(tableName, databaseNameOp)) match {
- case Some(parts) => new util.ArrayList(parts.toList.asJava)
- case _ => null
- }
}
operationContext.setProperty("isOverwrite", isOverwriteTable)
if(CarbonUtil.hasAggregationDataMap(table)) {
@@ -149,7 +143,16 @@ case class CarbonLoadDataCommand(
val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
val carbonProperty: CarbonProperties = CarbonProperties.getInstance()
carbonProperty.addProperty("zookeeper.enable.lock", "false")
-
+ currPartitions = if (table.isHivePartitionTable) {
+ CarbonFilters.getCurrentPartitions(
+ sparkSession,
+ table) match {
+ case Some(parts) => new util.ArrayList(parts.toList.asJava)
+ case _ => null
+ }
+ } else {
+ null
+ }
// get the value of 'spark.executor.cores' from spark conf, default value is 1
val sparkExecutorCores = sparkSession.sparkContext.conf.get("spark.executor.cores", "1")
// get the value of 'carbon.number.of.cores.while.loading' from carbon properties,
@@ -663,19 +666,6 @@ case class CarbonLoadDataCommand(
persistedRDD = persistedRDDLocal
transformedPlan
} else {
- val rowDataTypes = attributes.map { attribute =>
- catalogTable.schema.find(_.name.equalsIgnoreCase(attribute.name)) match {
- case Some(attr) => attr.dataType
- case _ => StringType
- }
- }
- // Find the partition columns from the csv header attributes
- val partitionColumns = attributes.map { attribute =>
- catalogTable.partitionSchema.find(_.name.equalsIgnoreCase(attribute.name)) match {
- case Some(attr) => true
- case _ => false
- }
- }
val columnCount = carbonLoadModel.getCsvHeaderColumns.length
val rdd = CsvRDDHelper.csvFileScanRDD(
sparkSession,
@@ -729,6 +719,7 @@ case class CarbonLoadDataCommand(
query = logicalPlan,
overwrite = false,
ifPartitionNotExists = false)
+ sparkSession.sparkContext.setLocalProperty(EXECUTION_ID_KEY, null)
Dataset.ofRows(sparkSession, convertedPlan)
} catch {
case ex: Throwable =>
@@ -768,10 +759,6 @@ case class CarbonLoadDataCommand(
table,
operationContext)
- val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =
- new LoadTablePreStatusUpdateEvent(table.getCarbonTableIdentifier, carbonLoadModel)
- OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)
-
} catch {
case e: Exception =>
throw new Exception(
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableAddHivePartitionCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableAddHivePartitionCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableAddHivePartitionCommand.scala
index 25a0d8e..59ec71c 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableAddHivePartitionCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableAddHivePartitionCommand.scala
@@ -55,6 +55,9 @@ case class CarbonAlterTableAddHivePartitionCommand(
override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
table = CarbonEnv.getCarbonTable(tableName)(sparkSession)
if (table.isHivePartitionTable) {
+ if (table.isChildDataMap) {
+ throw new UnsupportedOperationException("Cannot add partition directly on aggregate tables")
+ }
val partitionWithLoc = partitionSpecsAndLocs.filter(_._2.isDefined)
if (partitionWithLoc.nonEmpty) {
val partitionSpecs = partitionWithLoc.map{ case (part, location) =>
@@ -70,7 +73,7 @@ case class CarbonAlterTableAddHivePartitionCommand(
}.asJava)
}
val operationContext = new OperationContext
- val preAlterTableHivePartitionCommandEvent = new PreAlterTableHivePartitionCommandEvent(
+ val preAlterTableHivePartitionCommandEvent = PreAlterTableHivePartitionCommandEvent(
sparkSession,
table)
OperationListenerBus.getInstance()
@@ -81,6 +84,9 @@ case class CarbonAlterTableAddHivePartitionCommand(
table)
OperationListenerBus.getInstance()
.fireEvent(postAlterTableHivePartitionCommandEvent, operationContext)
+ } else {
+ throw new UnsupportedOperationException(
+ "Cannot add partition directly on non partitioned table")
}
Seq.empty[Row]
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/55084602/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
index c67d694..45cacfa 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/partition/CarbonAlterTableDropHivePartitionCommand.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.command.partition
import java.util
+import java.util.UUID
import scala.collection.JavaConverters._
@@ -35,7 +36,7 @@ import org.apache.carbondata.core.metadata.SegmentFileStore
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.CarbonUtil
-import org.apache.carbondata.events.{OperationContext, OperationListenerBus, PostAlterTableHivePartitionCommandEvent, PreAlterTableHivePartitionCommandEvent}
+import org.apache.carbondata.events._
import org.apache.carbondata.spark.rdd.CarbonDropPartitionRDD
/**
@@ -57,18 +58,16 @@ case class CarbonAlterTableDropHivePartitionCommand(
specs: Seq[TablePartitionSpec],
ifExists: Boolean,
purge: Boolean,
- retainData: Boolean)
+ retainData: Boolean,
+ operationContext: OperationContext = new OperationContext)
extends AtomicRunnableCommand {
var carbonPartitionsTobeDropped : util.List[PartitionSpec] = _
var table: CarbonTable = _
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
table = CarbonEnv.getCarbonTable(tableName)(sparkSession)
- if (CarbonUtil.hasAggregationDataMap(table)) {
- throw new AnalysisException(
- "Partition can not be dropped as it is mapped to Pre Aggregate table")
- }
if (table.isHivePartitionTable) {
var locks = List.empty[ICarbonLock]
try {
@@ -90,12 +89,14 @@ case class CarbonAlterTableDropHivePartitionCommand(
partition.location)
}
carbonPartitionsTobeDropped = new util.ArrayList[PartitionSpec](carbonPartitions.asJava)
- val operationContext = new OperationContext
val preAlterTableHivePartitionCommandEvent = PreAlterTableHivePartitionCommandEvent(
sparkSession,
table)
OperationListenerBus.getInstance()
.fireEvent(preAlterTableHivePartitionCommandEvent, operationContext)
+ val metaEvent = AlterTableDropPartitionMetaEvent(table, specs, ifExists, purge, retainData)
+ OperationListenerBus.getInstance()
+ .fireEvent(metaEvent, operationContext)
// Drop the partitions from hive.
AlterTableDropPartitionCommand(
tableName,
@@ -120,6 +121,10 @@ case class CarbonAlterTableDropHivePartitionCommand(
AlterTableUtil.releaseLocks(locks)
}
+ } else {
+ throwMetadataException(tableName.database.getOrElse(sparkSession.catalog.currentDatabase),
+ tableName.table,
+ "Not a partitioned table")
}
Seq.empty[Row]
}
@@ -129,7 +134,7 @@ case class CarbonAlterTableDropHivePartitionCommand(
AlterTableAddPartitionCommand(tableName, specs.map((_, None)), true)
val msg = s"Got exception $exception when processing data of drop partition." +
"Adding back partitions to the metadata"
- LogServiceFactory.getLogService(this.getClass.getCanonicalName).error(msg)
+ LOGGER.error(msg)
Seq.empty[Row]
}
@@ -147,9 +152,24 @@ case class CarbonAlterTableDropHivePartitionCommand(
table.getDatabaseName,
table.getTableName,
locksToBeAcquired)(sparkSession)
- val partitionNames = specs.flatMap { f =>
- f.map(k => k._1 + "=" + k._2)
- }.toSet
+ // If flow is for child table then get the uuid from operation context.
+ // If flow is for parent table then generate uuid for child flows and set the uuid to ""
+ // for parent table
+ // If normal table then set uuid to "".
+ val uuid = if (table.isChildDataMap) {
+ val uuid = operationContext.getProperty("uuid")
+ if (uuid != null) {
+ uuid.toString
+ } else {
+ LOGGER.warn(s"UUID not set for table ${table.getTableUniqueName} in operation context.")
+ ""
+ }
+ } else if (table.hasAggregationDataMap) {
+ operationContext.setProperty("uuid", UUID.randomUUID().toString)
+ ""
+ } else {
+ ""
+ }
val segments = new SegmentStatusManager(table.getAbsoluteTableIdentifier)
.getValidAndInvalidSegments.getValidSegments
// First drop the partitions from partition mapper files of each segment
@@ -168,7 +188,14 @@ case class CarbonAlterTableDropHivePartitionCommand(
tobeDeletedSegs.add(tobeDeleted.split(",")(0))
}
}
- SegmentFileStore.commitDropPartitions(table, uniqueId, tobeUpdatedSegs, tobeDeletedSegs)
+ val preStatusEvent = AlterTableDropPartitionPreStatusEvent(table)
+ OperationListenerBus.getInstance().fireEvent(preStatusEvent, operationContext)
+
+ SegmentFileStore.commitDropPartitions(table, uniqueId, tobeUpdatedSegs, tobeDeletedSegs, uuid)
+
+ val postStatusEvent = AlterTableDropPartitionPostStatusEvent(table)
+ OperationListenerBus.getInstance().fireEvent(postStatusEvent, operationContext)
+
DataMapStoreManager.getInstance().clearDataMaps(table.getAbsoluteTableIdentifier)
} finally {
AlterTableUtil.releaseLocks(locks)