You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by qi...@apache.org on 2017/11/29 10:41:15 UTC
[4/4] carbondata git commit: [CARBONDATA-1815][PreAgg] Add
AtomicRunnableCommand abstraction
[CARBONDATA-1815][PreAgg] Add AtomicRunnableCommand abstraction
This closes #1573
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/90fb6baf
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/90fb6baf
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/90fb6baf
Branch: refs/heads/master
Commit: 90fb6baf2736d9c1e6572ed29bebcc2801d793c5
Parents: aa9014a
Author: Jacky Li <ja...@qq.com>
Authored: Mon Nov 27 16:48:09 2017 +0800
Committer: QiangCai <qi...@qq.com>
Committed: Wed Nov 29 18:40:30 2017 +0800
----------------------------------------------------------------------
.../spark/sql/CarbonCatalystOperators.scala | 19 -
.../spark/sql/CarbonDataFrameWriter.scala | 4 +-
.../sql/CarbonDatasourceHadoopRelation.scala | 4 +-
.../scala/org/apache/spark/sql/CarbonEnv.scala | 49 ++-
.../command/CarbonCreateTableCommand.scala | 124 ------
.../CarbonDescribeFormattedCommand.scala | 152 -------
.../command/CarbonDropTableCommand.scala | 147 -------
.../datamap/CarbonCreateDataMapCommand.scala | 49 ++-
.../datamap/CarbonDataMapShowCommand.scala | 9 +-
.../datamap/CarbonDropDataMapCommand.scala | 16 +-
.../command/datamap/CreateDataMapCommand.scala | 69 ----
.../command/datamap/DataMapListeners.scala | 30 +-
.../AlterTableCompactionCommand.scala | 219 ----------
.../CarbonAlterTableCompactionCommand.scala | 215 ++++++++++
.../management/CarbonCleanFilesCommand.scala | 104 +++++
.../CarbonDeleteLoadByIdCommand.scala | 57 +++
.../CarbonDeleteLoadByLoadDateCommand.scala | 56 +++
.../management/CarbonInsertIntoCommand.scala | 49 +++
.../management/CarbonLoadDataCommand.scala | 410 +++++++++++++++++++
.../management/CarbonShowLoadsCommand.scala | 20 +-
.../command/management/CleanFilesCommand.scala | 108 -----
.../management/DeleteLoadByIdCommand.scala | 61 ---
.../DeleteLoadByLoadDateCommand.scala | 61 ---
.../management/LoadTableByInsertCommand.scala | 53 ---
.../command/management/LoadTableCommand.scala | 410 -------------------
.../CarbonProjectForDeleteCommand.scala | 113 +++++
.../CarbonProjectForUpdateCommand.scala | 223 ++++++++++
.../command/mutation/HorizontalCompaction.scala | 4 +-
.../mutation/ProjectForDeleteCommand.scala | 117 ------
.../mutation/ProjectForUpdateCommand.scala | 226 ----------
.../spark/sql/execution/command/package.scala | 61 ++-
.../AlterTableDropCarbonPartitionCommand.scala | 283 -------------
.../AlterTableSplitCarbonPartitionCommand.scala | 318 --------------
.../CarbonAlterTableDropPartitionCommand.scala | 277 +++++++++++++
.../CarbonAlterTableSplitPartitionCommand.scala | 312 ++++++++++++++
.../CarbonShowCarbonPartitionsCommand.scala | 55 +++
.../partition/ShowCarbonPartitionsCommand.scala | 59 ---
.../CreatePreAggregateTableCommand.scala | 105 +++--
.../preaaggregate/PreAggregateListeners.scala | 2 +-
.../preaaggregate/PreAggregateUtil.scala | 4 -
.../command/schema/AlterTableSetCommand.scala | 43 --
.../command/schema/AlterTableUnsetCommand.scala | 44 --
.../CarbonAlterTableAddColumnCommand.scala | 8 +-
.../CarbonAlterTableDataTypeChangeCommand.scala | 7 +-
.../CarbonAlterTableDropColumnCommand.scala | 14 +-
.../schema/CarbonAlterTableRenameCommand.scala | 6 +-
.../schema/CarbonAlterTableSetCommand.scala | 43 ++
.../schema/CarbonAlterTableUnsetCommand.scala | 44 ++
.../schema/CarbonGetTableDetailCommand.scala | 12 +-
.../table/CarbonCreateTableCommand.scala | 120 ++++++
.../table/CarbonDescribeFormattedCommand.scala | 148 +++++++
.../command/table/CarbonDropTableCommand.scala | 143 +++++++
.../sql/execution/strategy/DDLStrategy.scala | 25 +-
.../strategy/StreamingTableStrategy.scala | 6 +-
.../spark/sql/hive/CarbonAnalysisRules.scala | 4 +-
.../execution/command/CarbonHiveCommands.scala | 1 +
.../sql/optimizer/CarbonLateDecodeRule.scala | 4 +-
.../sql/parser/CarbonSpark2SqlParser.scala | 49 +--
.../spark/sql/parser/CarbonSparkSqlParser.scala | 3 +-
.../org/apache/spark/util/Compaction.scala | 4 +-
.../org/apache/spark/util/TableLoader.scala | 4 +-
.../vectorreader/VectorReaderTestCase.scala | 2 +-
62 files changed, 2681 insertions(+), 2707 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
index 79213d3..4f2cf14 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonCatalystOperators.scala
@@ -112,25 +112,6 @@ object GetDB {
}
}
-/**
- * Shows Loads in a table
- */
-case class ShowLoadsCommand(
- databaseNameOp: Option[String],
- table: String,
- limit: Option[String])
- extends Command {
-
- override def output: Seq[Attribute] = {
- Seq(AttributeReference("SegmentSequenceId", StringType, nullable = false)(),
- AttributeReference("Status", StringType, nullable = false)(),
- AttributeReference("Load Start Time", TimestampType, nullable = false)(),
- AttributeReference("Load End Time", TimestampType, nullable = true)(),
- AttributeReference("Merged To", StringType, nullable = false)(),
- AttributeReference("File Format", StringType, nullable = false)())
- }
-}
-
case class ProjectForUpdate(
table: UnresolvedRelation,
columns: List[String],
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
index bae93bc..71590dd 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
@@ -19,7 +19,7 @@ package org.apache.spark.sql
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.compress.GzipCodec
-import org.apache.spark.sql.execution.command.management.LoadTableCommand
+import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CarbonException
@@ -132,7 +132,7 @@ class CarbonDataFrameWriter(sqlContext: SQLContext, val dataFrame: DataFrame) {
*/
private def loadDataFrame(options: CarbonOption): Unit = {
val header = dataFrame.columns.mkString(",")
- LoadTableCommand(
+ CarbonLoadDataCommand(
Some(options.dbName),
options.tableName,
null,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
index 1a1fe33..70645f6 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
@@ -22,7 +22,7 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.spark.CarbonInputMetrics
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.execution.command.management.LoadTableByInsertCommand
+import org.apache.spark.sql.execution.command.management.CarbonInsertIntoCommand
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.sql.sources.{BaseRelation, Filter, InsertableRelation}
@@ -96,7 +96,7 @@ case class CarbonDatasourceHadoopRelation(
CarbonCommonConstants.DEFAULT_MAX_NUMBER_OF_COLUMNS)
}
if (data.logicalPlan.output.size >= carbonRelation.output.size) {
- LoadTableByInsertCommand(this, data.logicalPlan, overwrite).run(sparkSession)
+ CarbonInsertIntoCommand(this, data.logicalPlan, overwrite).run(sparkSession)
} else {
CarbonException.analysisException(
"Cannot insert into target table because number of columns mismatch")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/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 592e2c9..c03b210 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
@@ -25,9 +25,10 @@ import org.apache.spark.sql.hive.{CarbonMetaStore, CarbonMetaStoreFactory, Carbo
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util._
-import org.apache.carbondata.events.{CarbonEnvInitPreEvent, OperationListenerBus}
+import org.apache.carbondata.core.util.path.{CarbonStorePath, CarbonTablePath}
import org.apache.carbondata.events.{CarbonEnvInitPreEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.spark.rdd.SparkReadSupport
import org.apache.carbondata.spark.readsupport.SparkRowReadSupportImpl
@@ -68,7 +69,7 @@ class CarbonEnv {
carbonSessionInfo.setSessionParams(sessionParams)
ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)
val config = new CarbonSQLConf(sparkSession)
- if (sparkSession.conf.getOption(CarbonCommonConstants.ENABLE_UNSAFE_SORT) == None) {
+ if (sparkSession.conf.getOption(CarbonCommonConstants.ENABLE_UNSAFE_SORT).isEmpty) {
config.addDefaultCarbonParams()
}
// add session params after adding DefaultCarbonParams
@@ -116,7 +117,7 @@ object CarbonEnv {
}
/**
- * Return carbon table instance by looking up relation in `sparkSession`
+ * Return carbon table instance by looking up table in `sparkSession`
*/
def getCarbonTable(
databaseNameOp: Option[String],
@@ -130,6 +131,9 @@ object CarbonEnv {
.carbonTable
}
+ /**
+ * Return carbon table instance by looking up table in `sparkSession`
+ */
def getCarbonTable(
tableIdentifier: TableIdentifier)
(sparkSession: SparkSession): CarbonTable = {
@@ -140,4 +144,43 @@ object CarbonEnv {
.asInstanceOf[CarbonRelation]
.carbonTable
}
+
+ /**
+ * Return database name or get default name from sparkSession
+ */
+ def getDatabaseName(
+ databaseNameOp: Option[String]
+ )(sparkSession: SparkSession): String = {
+ databaseNameOp.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase)
+ }
+
+ /**
+ * Return table path for specified table
+ */
+ def getTablePath(
+ databaseNameOp: Option[String],
+ tableName: String
+ )(sparkSession: SparkSession): String = {
+ val dbLocation = GetDB.getDatabaseLocation(
+ databaseNameOp.getOrElse(sparkSession.sessionState.catalog.getCurrentDatabase),
+ sparkSession,
+ CarbonProperties.getStorePath)
+ dbLocation + CarbonCommonConstants.FILE_SEPARATOR + tableName
+ }
+
+ /**
+ * Return metadata path for specified table
+ */
+ def getMetadataPath(
+ databaseNameOp: Option[String],
+ tableName: String
+ )(sparkSession: SparkSession): String = {
+ val absoluteTableIdentifier = AbsoluteTableIdentifier.from(
+ getTablePath(databaseNameOp, tableName)(sparkSession),
+ getDatabaseName(databaseNameOp)(sparkSession),
+ tableName)
+ val carbonTablePath = CarbonStorePath.getCarbonTablePath(absoluteTableIdentifier)
+ val schemaFilePath = carbonTablePath.getSchemaFilePath
+ CarbonTablePath.getFolderContainingFile(schemaFilePath)
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonCreateTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonCreateTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonCreateTableCommand.scala
deleted file mode 100644
index f9b8b9e..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonCreateTableCommand.scala
+++ /dev/null
@@ -1,124 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.command
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.sql._
-import org.apache.spark.sql.{CarbonEnv, GetDB, Row, SparkSession}
-import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
-import org.apache.spark.sql.execution.SQLExecution.EXECUTION_ID_KEY
-import org.apache.spark.sql.util.CarbonException
-
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.exception.InvalidConfigurationException
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
-import org.apache.carbondata.core.metadata.schema.table.TableInfo
-import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
-import org.apache.carbondata.events.{CreateTablePostExecutionEvent, CreateTablePreExecutionEvent, OperationContext, OperationListenerBus}
-
-case class CarbonCreateTableCommand(
- cm: TableModel,
- createDSTable: Boolean = true)
- extends RunnableCommand with SchemaProcessCommand {
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processSchema(sparkSession)
- }
-
- override def processSchema(sparkSession: SparkSession): Seq[Row] = {
- val storePath = CarbonProperties.getStorePath
- CarbonEnv.getInstance(sparkSession).carbonMetastore.
- checkSchemasModifiedTimeAndReloadTables()
- val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
- cm.databaseName = GetDB.getDatabaseName(cm.databaseNameOp, sparkSession)
- val dbLocation = GetDB.getDatabaseLocation(cm.databaseName, sparkSession, storePath)
- val tablePath = dbLocation + CarbonCommonConstants.FILE_SEPARATOR + cm.tableName
- val tbName = cm.tableName
- val dbName = cm.databaseName
- LOGGER.audit(s"Creating Table with Database name [$dbName] and Table name [$tbName]")
-
- val tableInfo: TableInfo = TableNewProcessor(cm)
-
- // Add validation for sort scope when create table
- val sortScope = tableInfo.getFactTable.getTableProperties.asScala
- .getOrElse("sort_scope", CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT)
- if (!CarbonUtil.isValidSortOption(sortScope)) {
- throw new InvalidConfigurationException(
- s"Passing invalid SORT_SCOPE '$sortScope', valid SORT_SCOPE are 'NO_SORT', 'BATCH_SORT'," +
- s" 'LOCAL_SORT' and 'GLOBAL_SORT' ")
- }
-
- if (tableInfo.getFactTable.getListOfColumns.size <= 0) {
- CarbonException.analysisException("Table should have at least one column.")
- }
-
- if (sparkSession.sessionState.catalog.listTables(dbName)
- .exists(_.table.equalsIgnoreCase(tbName))) {
- if (!cm.ifNotExistsSet) {
- LOGGER.audit(
- s"Table creation with Database name [$dbName] and Table name [$tbName] failed. " +
- s"Table [$tbName] already exists under database [$dbName]")
- throw new TableAlreadyExistsException(dbName, dbName)
- }
- } else {
- val tableIdentifier = AbsoluteTableIdentifier.from(tablePath, dbName, tbName)
- val operationContext = new OperationContext
- val createTablePreExecutionEvent: CreateTablePreExecutionEvent =
- new CreateTablePreExecutionEvent(sparkSession,
- tableIdentifier.getCarbonTableIdentifier,
- tablePath)
- OperationListenerBus.getInstance.fireEvent(createTablePreExecutionEvent, operationContext)
- // Add Database to catalog and persist
- val catalog = CarbonEnv.getInstance(sparkSession).carbonMetastore
- val carbonSchemaString = catalog.generateTableSchemaString(tableInfo, tableIdentifier)
- if (createDSTable) {
- try {
- val fields = new Array[Field](cm.dimCols.size + cm.msrCols.size)
- cm.dimCols.foreach(f => fields(f.schemaOrdinal) = f)
- cm.msrCols.foreach(f => fields(f.schemaOrdinal) = f)
-
- sparkSession.sparkContext.setLocalProperty(EXECUTION_ID_KEY, null)
- sparkSession.sql(
- s"""CREATE TABLE $dbName.$tbName
- |(${ fields.map(f => f.rawSchema).mkString(",") })
- |USING org.apache.spark.sql.CarbonSource""".stripMargin +
- s""" OPTIONS (tableName "$tbName", dbName "$dbName", tablePath """.stripMargin +
- s""""$tablePath", path "$tablePath" $carbonSchemaString) """.stripMargin)
- } catch {
- case e: AnalysisException => throw e
- case e: Exception =>
- // call the drop table to delete the created table.
- CarbonEnv.getInstance(sparkSession).carbonMetastore
- .dropTable(tableIdentifier)(sparkSession)
-
- val msg = s"Create table'$tbName' in database '$dbName' failed."
- LOGGER.audit(msg)
- LOGGER.error(e, msg)
- CarbonException.analysisException(msg)
- }
- }
- val createTablePostExecutionEvent: CreateTablePostExecutionEvent =
- new CreateTablePostExecutionEvent(sparkSession, tableIdentifier.getCarbonTableIdentifier)
- OperationListenerBus.getInstance.fireEvent(createTablePostExecutionEvent, operationContext)
- LOGGER.audit(s"Table created with Database name [$dbName] and Table name [$tbName]")
- }
- Seq.empty
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDescribeFormattedCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDescribeFormattedCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDescribeFormattedCommand.scala
deleted file mode 100644
index b9ba9cb..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDescribeFormattedCommand.scala
+++ /dev/null
@@ -1,152 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.command
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
-import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.expressions.Attribute
-import org.apache.spark.sql.execution.SparkPlan
-import org.apache.spark.sql.hive.CarbonRelation
-import org.codehaus.jackson.map.ObjectMapper
-
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.metadata.encoder.Encoding
-import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.core.util.CarbonUtil
-
-private[sql] case class CarbonDescribeFormattedCommand(
- child: SparkPlan,
- override val output: Seq[Attribute],
- tblIdentifier: TableIdentifier)
- extends RunnableCommand with SchemaProcessCommand {
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processSchema(sparkSession)
- }
-
- private def getColumnGroups(dimensions: List[CarbonDimension]): Seq[(String, String, String)] = {
- var results: Seq[(String, String, String)] =
- Seq(("", "", ""), ("##Column Group Information", "", ""))
- val groupedDimensions = dimensions.groupBy(x => x.columnGroupId()).filter {
- case (groupId, _) => groupId != -1
- }.toSeq.sortBy(_._1)
- val groups = groupedDimensions.map(colGroups => {
- colGroups._2.map(dim => dim.getColName).mkString(", ")
- })
- var index = 1
- groups.foreach { x =>
- results = results :+ (s"Column Group $index", x, "")
- index = index + 1
- }
- results
- }
-
- override def processSchema(sparkSession: SparkSession): Seq[Row] = {
- val relation = CarbonEnv.getInstance(sparkSession).carbonMetastore
- .lookupRelation(tblIdentifier)(sparkSession).asInstanceOf[CarbonRelation]
- val mapper = new ObjectMapper()
- val colProps = StringBuilder.newBuilder
- val dims = relation.metaData.dims.map(x => x.toLowerCase)
- var results: Seq[(String, String, String)] = child.schema.fields.map { field =>
- val fieldName = field.name.toLowerCase
- val colComment = field.getComment().getOrElse("null")
- val comment = if (dims.contains(fieldName)) {
- val dimension = relation.metaData.carbonTable.getDimensionByName(
- relation.carbonTable.getTableName, fieldName)
- if (null != dimension.getColumnProperties && !dimension.getColumnProperties.isEmpty) {
- colProps.append(fieldName).append(".")
- .append(mapper.writeValueAsString(dimension.getColumnProperties))
- .append(",")
- }
- if (dimension.hasEncoding(Encoding.DICTIONARY) &&
- !dimension.hasEncoding(Encoding.DIRECT_DICTIONARY)) {
- "DICTIONARY, KEY COLUMN" + (if (dimension.hasEncoding(Encoding.INVERTED_INDEX)) {
- "".concat(",").concat(colComment)
- } else {
- ",NOINVERTEDINDEX".concat(",").concat(colComment)
- })
- } else {
- "KEY COLUMN" + (if (dimension.hasEncoding(Encoding.INVERTED_INDEX)) {
- "".concat(",").concat(colComment)
- } else {
- ",NOINVERTEDINDEX".concat(",").concat(colComment)
- })
- }
- } else {
- "MEASURE".concat(",").concat(colComment)
- }
-
- (field.name, field.dataType.simpleString, comment)
- }
- val colPropStr = if (colProps.toString().trim().length() > 0) {
- // drops additional comma at end
- colProps.toString().dropRight(1)
- } else {
- colProps.toString()
- }
- results ++= Seq(("", "", ""), ("##Detailed Table Information", "", ""))
- results ++= Seq(("Database Name: ", relation.carbonTable.getDatabaseName, "")
- )
- results ++= Seq(("Table Name: ", relation.carbonTable.getTableName, ""))
- results ++= Seq(("CARBON Store Path: ", CarbonProperties.getStorePath, ""))
- val carbonTable = relation.carbonTable
- // Carbon table support table comment
- val tableComment = carbonTable.getTableInfo.getFactTable.getTableProperties.asScala
- .getOrElse(CarbonCommonConstants.TABLE_COMMENT, "")
- results ++= Seq(("Comment: ", tableComment, ""))
- results ++= Seq(("Table Block Size : ", carbonTable.getBlockSizeInMB + " MB", ""))
- val dataIndexSize = CarbonUtil.calculateDataIndexSize(carbonTable)
- if (!dataIndexSize.isEmpty) {
- results ++= Seq((CarbonCommonConstants.TABLE_DATA_SIZE + ":",
- dataIndexSize.get(CarbonCommonConstants.CARBON_TOTAL_DATA_SIZE).toString, ""))
- results ++= Seq((CarbonCommonConstants.TABLE_INDEX_SIZE + ":",
- dataIndexSize.get(CarbonCommonConstants.CARBON_TOTAL_INDEX_SIZE).toString, ""))
- results ++= Seq((CarbonCommonConstants.LAST_UPDATE_TIME + ":",
- dataIndexSize.get(CarbonCommonConstants.LAST_UPDATE_TIME).toString, ""))
- }
- results ++= Seq(("SORT_SCOPE", carbonTable.getTableInfo.getFactTable
- .getTableProperties.asScala.getOrElse("sort_scope", CarbonCommonConstants
- .LOAD_SORT_SCOPE_DEFAULT), CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT))
- results ++= Seq(("", "", ""), ("##Detailed Column property", "", ""))
- if (colPropStr.length() > 0) {
- results ++= Seq((colPropStr, "", ""))
- } else {
- results ++= Seq(("ADAPTIVE", "", ""))
- }
- results ++= Seq(("SORT_COLUMNS", relation.metaData.carbonTable.getSortColumns(
- relation.carbonTable.getTableName).asScala
- .map(column => column).mkString(","), ""))
- val dimension = carbonTable
- .getDimensionByTableName(relation.carbonTable.getTableName)
- results ++= getColumnGroups(dimension.asScala.toList)
- if (carbonTable.getPartitionInfo(carbonTable.getTableName) != null) {
- results ++=
- Seq(("Partition Columns: ", carbonTable.getPartitionInfo(carbonTable.getTableName)
- .getColumnSchemaList.asScala.map(_.getColumnName).mkString(","), ""))
- }
- results.map {
- case (name, dataType, null) =>
- Row(f"$name%-36s", f"$dataType%-80s", null)
- case (name, dataType, comment) =>
- Row(f"$name%-36s", f"$dataType%-80s", f"$comment%-72s")
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDropTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDropTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDropTableCommand.scala
deleted file mode 100644
index ddd5a76..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/CarbonDropTableCommand.scala
+++ /dev/null
@@ -1,147 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.command
-
-import scala.collection.mutable.ListBuffer
-
-import org.apache.spark.sql.{CarbonEnv, GetDB, Row, SparkSession}
-import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
-import org.apache.spark.sql.hive.CarbonRelation
-import org.apache.spark.sql.util.CarbonException
-
-import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.datastore.impl.FileFactory
-import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage}
-import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
-import org.apache.carbondata.core.metadata.schema.table.CarbonTable
-import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
-import org.apache.carbondata.core.util.path.CarbonStorePath
-import org.apache.carbondata.events._
-
-case class CarbonDropTableCommand(
- ifExistsSet: Boolean,
- databaseNameOp: Option[String],
- tableName: String,
- dropChildTable: Boolean = false)
- extends RunnableCommand with SchemaProcessCommand with DataProcessCommand {
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processSchema(sparkSession)
- processData(sparkSession)
- }
-
- override def processSchema(sparkSession: SparkSession): Seq[Row] = {
- val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
- val dbName = GetDB.getDatabaseName(databaseNameOp, sparkSession)
- val identifier = TableIdentifier(tableName, Option(dbName))
- val locksToBeAcquired = List(LockUsage.METADATA_LOCK, LockUsage.DROP_TABLE_LOCK)
- val carbonEnv = CarbonEnv.getInstance(sparkSession)
- val catalog = carbonEnv.carbonMetastore
- val databaseLocation = GetDB.getDatabaseLocation(dbName, sparkSession,
- CarbonProperties.getStorePath)
- val tablePath = databaseLocation + CarbonCommonConstants.FILE_SEPARATOR + tableName.toLowerCase
- val absoluteTableIdentifier = AbsoluteTableIdentifier
- .from(tablePath, dbName.toLowerCase, tableName.toLowerCase)
- catalog.checkSchemasModifiedTimeAndReloadTables()
- val carbonLocks: scala.collection.mutable.ListBuffer[ICarbonLock] = ListBuffer()
- try {
- locksToBeAcquired foreach {
- lock => carbonLocks += CarbonLockUtil.getLockObject(absoluteTableIdentifier, lock)
- }
- LOGGER.audit(s"Deleting table [$tableName] under database [$dbName]")
- val carbonTable: Option[CarbonTable] =
- catalog.getTableFromMetadataCache(dbName, tableName) match {
- case Some(carbonTable) => Some(carbonTable)
- case None => try {
- Some(catalog.lookupRelation(identifier)(sparkSession)
- .asInstanceOf[CarbonRelation].metaData.carbonTable)
- } catch {
- case ex: NoSuchTableException =>
- if (!ifExistsSet) {
- throw ex
- }
- None
- }
- }
- if (carbonTable.isDefined) {
- val relationIdentifiers = carbonTable.get.getTableInfo.getParentRelationIdentifiers
- if (relationIdentifiers != null && !relationIdentifiers.isEmpty) {
- if (!dropChildTable) {
- if (!ifExistsSet) {
- throw new Exception("Child table which is associated with datamap cannot " +
- "be dropped, use DROP DATAMAP command to drop")
- } else {
- return Seq.empty
- }
- }
- }
- }
- val operationContext = new OperationContext
- val dropTablePreEvent: DropTablePreEvent =
- DropTablePreEvent(
- carbonTable,
- ifExistsSet,
- sparkSession)
- OperationListenerBus.getInstance.fireEvent(dropTablePreEvent, operationContext)
- CarbonEnv.getInstance(sparkSession).carbonMetastore
- .dropTable(absoluteTableIdentifier)(sparkSession)
-
- // fires the event after dropping main table
- val dropTablePostEvent: DropTablePostEvent =
- DropTablePostEvent(
- carbonTable,
- ifExistsSet,
- sparkSession)
- OperationListenerBus.getInstance.fireEvent(dropTablePostEvent, operationContext)
- LOGGER.audit(s"Deleted table [$tableName] under database [$dbName]")
- } catch {
- case ex: Exception =>
- LOGGER.error(ex, s"Dropping table $dbName.$tableName failed")
- CarbonException.analysisException(
- s"Dropping table $dbName.$tableName failed: ${ ex.getMessage }")
- }
- finally {
- if (carbonLocks.nonEmpty) {
- val unlocked = carbonLocks.forall(_.unlock())
- if (unlocked) {
- logInfo("Table MetaData Unlocked Successfully")
- }
- }
- }
- Seq.empty
- }
-
- override def processData(sparkSession: SparkSession): Seq[Row] = {
- // delete the table folder
- val dbName = GetDB.getDatabaseName(databaseNameOp, sparkSession)
- val databaseLocation = GetDB.getDatabaseLocation(dbName, sparkSession,
- CarbonProperties.getStorePath)
- val tablePath = databaseLocation + CarbonCommonConstants.FILE_SEPARATOR + tableName.toLowerCase
- val tableIdentifier = AbsoluteTableIdentifier.from(tablePath, dbName, tableName)
- val metadataFilePath =
- CarbonStorePath.getCarbonTablePath(tableIdentifier).getMetadataDirectoryPath
- val fileType = FileFactory.getFileType(metadataFilePath)
- if (FileFactory.isFileExist(metadataFilePath, fileType)) {
- val file = FileFactory.getCarbonFile(metadataFilePath, fileType)
- CarbonUtil.deleteFoldersAndFilesSilent(file.getParentFile)
- }
- Seq.empty
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
index 8b6e3b9..29c60ea 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
@@ -29,8 +29,6 @@ import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
/**
* Below command class will be used to create datamap on table
* and updating the parent table about the datamap information
- *
- * @param queryString
*/
case class CarbonCreateDataMapCommand(
dataMapName: String,
@@ -38,21 +36,19 @@ case class CarbonCreateDataMapCommand(
dmClassName: String,
dmproperties: Map[String, String],
queryString: Option[String])
- extends RunnableCommand with SchemaProcessCommand {
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processSchema(sparkSession)
- }
+ extends AtomicRunnableCommand {
- override def processSchema(sparkSession: SparkSession): Seq[Row] = {
+ override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
dmClassName.equalsIgnoreCase("preaggregate")) {
- CreatePreAggregateTableCommand(dataMapName,
+ CreatePreAggregateTableCommand(
+ dataMapName,
tableIdentifier,
dmClassName,
dmproperties,
- queryString.get).run(sparkSession)
+ queryString.get
+ ).processMetadata(sparkSession)
} else {
val dataMapSchema = new DataMapSchema(dataMapName, dmClassName)
dataMapSchema.setProperties(new java.util.HashMap[String, String](dmproperties.asJava))
@@ -60,9 +56,40 @@ case class CarbonCreateDataMapCommand(
// upadting the parent table about dataschema
PreAggregateUtil.updateMainTable(dbName, tableIdentifier.table, dataMapSchema, sparkSession)
}
- LOGGER.audit(s"DataMap ${dataMapName} successfully added to Table ${tableIdentifier.table}")
+ LOGGER.audit(s"DataMap $dataMapName successfully added to Table ${ tableIdentifier.table }")
Seq.empty
}
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
+ dmClassName.equalsIgnoreCase("preaggregate")) {
+ CreatePreAggregateTableCommand(
+ dataMapName,
+ tableIdentifier,
+ dmClassName,
+ dmproperties,
+ queryString.get
+ ).processData(sparkSession)
+ } else {
+ Seq.empty
+ }
+ }
+
+ override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
+ if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
+ dmClassName.equalsIgnoreCase("preaggregate")) {
+ CreatePreAggregateTableCommand(
+ dataMapName,
+ tableIdentifier,
+ dmClassName,
+ dmproperties,
+ queryString.get
+ ).undoMetadata(sparkSession, exception)
+ } else {
+ Seq.empty
+ }
+ }
+
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapShowCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapShowCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapShowCommand.scala
index 64a066c..fdd17d9 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapShowCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDataMapShowCommand.scala
@@ -21,8 +21,7 @@ import scala.collection.JavaConverters._
import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
-import org.apache.spark.sql.execution.command.{Checker, DataProcessCommand, RunnableCommand}
-import org.apache.spark.sql.hive.CarbonRelation
+import org.apache.spark.sql.execution.command.{Checker, DataCommand}
import org.apache.spark.sql.types.StringType
/**
@@ -33,7 +32,7 @@ import org.apache.spark.sql.types.StringType
case class CarbonDataMapShowCommand(
databaseNameOp: Option[String],
tableName: String)
- extends RunnableCommand with DataProcessCommand {
+ extends DataCommand {
override def output: Seq[Attribute] = {
Seq(AttributeReference("DataMapName", StringType, nullable = false)(),
@@ -41,10 +40,6 @@ case class CarbonDataMapShowCommand(
AttributeReference("Associated Table", StringType, nullable = false)())
}
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processData(sparkSession)
- }
-
override def processData(sparkSession: SparkSession): Seq[Row] = {
Checker.validateTableExists(databaseNameOp, tableName, sparkSession)
val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
index f34afbf..46c803d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonDropDataMapCommand.scala
@@ -23,7 +23,7 @@ import scala.collection.mutable.ListBuffer
import org.apache.spark.sql.{CarbonEnv, GetDB, Row, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
-import org.apache.spark.sql.execution.command.{DataProcessCommand, RunnableCommand, SchemaProcessCommand}
+import org.apache.spark.sql.execution.command.AtomicRunnableCommand
import org.apache.spark.sql.execution.command.preaaggregate.PreAggregateUtil
import org.apache.spark.sql.hive.CarbonRelation
@@ -31,7 +31,7 @@ import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage}
-import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.CarbonProperties
@@ -50,14 +50,9 @@ case class CarbonDropDataMapCommand(
ifExistsSet: Boolean,
databaseNameOp: Option[String],
tableName: String)
- extends RunnableCommand with SchemaProcessCommand with DataProcessCommand {
+ extends AtomicRunnableCommand {
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processSchema(sparkSession)
- processData(sparkSession)
- }
-
- override def processSchema(sparkSession: SparkSession): Seq[Row] = {
+ override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
val dbName = GetDB.getDatabaseName(databaseNameOp, sparkSession)
val identifier = TableIdentifier(tableName, Option(dbName))
@@ -132,7 +127,7 @@ case class CarbonDropDataMapCommand(
if (carbonLocks.nonEmpty) {
val unlocked = carbonLocks.forall(_.unlock())
if (unlocked) {
- logInfo("Table MetaData Unlocked Successfully")
+ LOGGER.info("Table MetaData Unlocked Successfully")
}
}
}
@@ -149,4 +144,5 @@ case class CarbonDropDataMapCommand(
DataMapStoreManager.getInstance().clearDataMap(tableIdentifier, dataMapName)
Seq.empty
}
+
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CreateDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CreateDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CreateDataMapCommand.scala
deleted file mode 100644
index abf461e..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CreateDataMapCommand.scala
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.spark.sql.execution.command.datamap
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.sql._
-import org.apache.spark.sql.catalyst.TableIdentifier
-import org.apache.spark.sql.execution.command._
-import org.apache.spark.sql.execution.command.preaaggregate.{CreatePreAggregateTableCommand, PreAggregateUtil}
-
-import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
-
-/**
- * Below command class will be used to create datamap on table
- * and updating the parent table about the datamap information
- *
- * @param queryString
- */
-case class CreateDataMapCommand(
- dataMapName: String,
- tableIdentifier: TableIdentifier,
- dmClassName: String,
- dmproperties: Map[String, String],
- queryString: Option[String])
- extends RunnableCommand with SchemaProcessCommand {
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processSchema(sparkSession)
- }
-
- override def processSchema(sparkSession: SparkSession): Seq[Row] = {
- val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
- if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
- dmClassName.equalsIgnoreCase("preaggregate")) {
- CreatePreAggregateTableCommand(dataMapName,
- tableIdentifier,
- dmClassName,
- dmproperties,
- queryString.get).run(sparkSession)
- } else {
-
- val dataMapSchema = new DataMapSchema(dataMapName, dmClassName)
- dataMapSchema.setProperties(new java.util.HashMap[String, String](dmproperties.asJava))
- val dbName = GetDB.getDatabaseName(tableIdentifier.database, sparkSession)
- // upadting the parent table about dataschema
- PreAggregateUtil.updateMainTable(dbName, tableIdentifier.table, dataMapSchema, sparkSession)
- }
- LOGGER.audit(s"DataMap ${dataMapName} successfully added to Table ${tableIdentifier.table}")
- Seq.empty
- }
-}
-
-
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
index a693d4c..45f99fd 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/DataMapListeners.scala
@@ -18,44 +18,41 @@ package org.apache.spark.sql.execution.command.datamap
import scala.collection.JavaConverters._
-import org.apache.spark.sql.execution.command.CarbonDropTableCommand
+import org.apache.spark.sql.execution.command.AtomicRunnableCommand
+import org.apache.spark.sql.execution.command.table.CarbonDropTableCommand
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
import org.apache.carbondata.events.{DropDataMapPostEvent, DropTablePostEvent, Event, OperationContext, OperationEventListener}
object DataMapDropTablePostListener extends OperationEventListener {
/**
- * Called on a specified event occurrence
- *
- * @param event
+ * Called on DropTablePostEvent occurrence
*/
override def onEvent(event: Event, operationContext: OperationContext): Unit = {
val dropPostEvent = event.asInstanceOf[DropTablePostEvent]
val carbonTable = dropPostEvent.carbonTable
val sparkSession = dropPostEvent.sparkSession
if (carbonTable.isDefined && carbonTable.get.hasDataMapSchema) {
+ // drop all child tables
val childSchemas = carbonTable.get.getTableInfo.getDataMapSchemaList
- for (childSchema: DataMapSchema <- childSchemas.asScala) {
- if (childSchema.getRelationIdentifier != null) {
+ childSchemas.asScala
+ .filter(_.getRelationIdentifier != null)
+ .foreach { childSchema =>
CarbonDropTableCommand(
ifExistsSet = true,
Some(childSchema.getRelationIdentifier.getDatabaseName),
- childSchema.getRelationIdentifier.getTableName, true
+ childSchema.getRelationIdentifier.getTableName,
+ dropChildTable = true
).run(sparkSession)
}
- }
}
-
}
}
object DropDataMapPostListener extends OperationEventListener {
/**
- * Called on a specified event occurrence
- *
- * @param event
+ * Called on DropDataMapPostEvent occurrence
*/
override def onEvent(event: Event, operationContext: OperationContext): Unit = {
val dropPostEvent = event.asInstanceOf[DropDataMapPostEvent]
@@ -63,9 +60,12 @@ object DropDataMapPostListener extends OperationEventListener {
val sparkSession = dropPostEvent.sparkSession
if (dataMapSchema.isDefined) {
if (dataMapSchema.get.getRelationIdentifier != null) {
- CarbonDropTableCommand(ifExistsSet = true,
+ CarbonDropTableCommand(
+ ifExistsSet = true,
Some(dataMapSchema.get.getRelationIdentifier.getDatabaseName),
- dataMapSchema.get.getRelationIdentifier.getTableName, true).run(sparkSession)
+ dataMapSchema.get.getRelationIdentifier.getTableName,
+ dropChildTable = true
+ ).run(sparkSession)
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/AlterTableCompactionCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/AlterTableCompactionCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/AlterTableCompactionCommand.scala
deleted file mode 100644
index cc398b5..0000000
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/AlterTableCompactionCommand.scala
+++ /dev/null
@@ -1,219 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.sql.execution.command.management
-
-import scala.collection.JavaConverters._
-
-import org.apache.spark.sql.{CarbonEnv, Row, SparkSession, SQLContext}
-import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
-import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, DataProcessCommand, RunnableCommand}
-import org.apache.spark.sql.hive.CarbonRelation
-import org.apache.spark.sql.util.CarbonException
-
-import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
-import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo}
-import org.apache.carbondata.core.mutate.CarbonUpdateUtil
-import org.apache.carbondata.core.util.CarbonProperties
-import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
-import org.apache.carbondata.processing.merger.{CarbonDataMergerUtil, CompactionType}
-import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory
-import org.apache.carbondata.spark.util.CommonUtil
-import org.apache.carbondata.streaming.StreamHandoffRDD
-
-/**
- * Command for the compaction in alter table command
- */
-case class AlterTableCompactionCommand(
- alterTableModel: AlterTableModel,
- tableInfoOp: Option[TableInfo] = None)
- extends RunnableCommand with DataProcessCommand {
-
- override def run(sparkSession: SparkSession): Seq[Row] = {
- processData(sparkSession)
- }
-
- override def processData(sparkSession: SparkSession): Seq[Row] = {
- val LOGGER: LogService =
- LogServiceFactory.getLogService(this.getClass.getName)
- val tableName = alterTableModel.tableName.toLowerCase
- val databaseName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
-
- val table = if (tableInfoOp.isDefined) {
- val tableInfo = tableInfoOp.get
- // To DO: CarbonEnv.updateStorePath
- CarbonTable.buildFromTableInfo(tableInfo)
- } else {
- val relation =
- CarbonEnv.getInstance(sparkSession).carbonMetastore
- .lookupRelation(Option(databaseName), tableName)(sparkSession)
- .asInstanceOf[CarbonRelation]
- if (relation == null) {
- throw new NoSuchTableException(databaseName, tableName)
- }
- if (null == relation.carbonTable) {
- LOGGER.error(s"alter table failed. table not found: $databaseName.$tableName")
- throw new NoSuchTableException(databaseName, tableName)
- }
- relation.carbonTable
- }
-
- val carbonLoadModel = new CarbonLoadModel()
- carbonLoadModel.setTableName(table.getTableName)
- val dataLoadSchema = new CarbonDataLoadSchema(table)
- // Need to fill dimension relation
- carbonLoadModel.setCarbonDataLoadSchema(dataLoadSchema)
- carbonLoadModel.setTableName(table.getTableName)
- carbonLoadModel.setDatabaseName(table.getDatabaseName)
- carbonLoadModel.setTablePath(table.getTablePath)
-
- var storeLocation = CarbonProperties.getInstance.getProperty(
- CarbonCommonConstants.STORE_LOCATION_TEMP_PATH,
- System.getProperty("java.io.tmpdir"))
- storeLocation = storeLocation + "/carbonstore/" + System.nanoTime()
- try {
- alterTableForCompaction(
- sparkSession.sqlContext,
- alterTableModel,
- carbonLoadModel,
- storeLocation)
- } catch {
- case e: Exception =>
- if (null != e.getMessage) {
- CarbonException.analysisException(
- s"Compaction failed. Please check logs for more info. ${ e.getMessage }")
- } else {
- CarbonException.analysisException(
- "Exception in compaction. Please check logs for more info.")
- }
- }
- Seq.empty
- }
-
- private def alterTableForCompaction(sqlContext: SQLContext,
- alterTableModel: AlterTableModel,
- carbonLoadModel: CarbonLoadModel,
- storeLocation: String): Unit = {
- val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getName)
- val compactionType = CompactionType.valueOf(alterTableModel.compactionType.toUpperCase)
- val compactionSize: Long = CarbonDataMergerUtil.getCompactionSize(compactionType)
- if (CompactionType.IUD_UPDDEL_DELTA == compactionType) {
- if (alterTableModel.segmentUpdateStatusManager.isDefined) {
- carbonLoadModel.setSegmentUpdateStatusManager(
- alterTableModel.segmentUpdateStatusManager.get)
- carbonLoadModel.setLoadMetadataDetails(
- alterTableModel.segmentUpdateStatusManager.get.getLoadMetadataDetails.toList.asJava)
- }
- }
-
- LOGGER.audit(s"Compaction request received for table " +
- s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
- val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
-
- if (null == carbonLoadModel.getLoadMetadataDetails) {
- CommonUtil.readLoadMetadataDetails(carbonLoadModel)
- }
-
- if (compactionType == CompactionType.STREAMING) {
- StreamHandoffRDD.startStreamingHandoffThread(
- carbonLoadModel,
- sqlContext,
- storeLocation)
- return
- }
-
- // reading the start time of data load.
- val loadStartTime : Long =
- if (alterTableModel.factTimeStamp.isEmpty) {
- CarbonUpdateUtil.readCurrentTime
- } else {
- alterTableModel.factTimeStamp.get
- }
- carbonLoadModel.setFactTimeStamp(loadStartTime)
-
- val isCompactionTriggerByDDl = true
- val compactionModel = CompactionModel(compactionSize,
- compactionType,
- carbonTable,
- isCompactionTriggerByDDl
- )
-
- val isConcurrentCompactionAllowed = CarbonProperties.getInstance()
- .getProperty(CarbonCommonConstants.ENABLE_CONCURRENT_COMPACTION,
- CarbonCommonConstants.DEFAULT_ENABLE_CONCURRENT_COMPACTION
- )
- .equalsIgnoreCase("true")
-
- // if system level compaction is enabled then only one compaction can run in the system
- // if any other request comes at this time then it will create a compaction request file.
- // so that this will be taken up by the compaction process which is executing.
- if (!isConcurrentCompactionAllowed) {
- LOGGER.info("System level compaction lock is enabled.")
- CarbonDataRDDFactory.handleCompactionForSystemLocking(
- sqlContext,
- carbonLoadModel,
- storeLocation,
- compactionType,
- carbonTable,
- compactionModel
- )
- } else {
- // normal flow of compaction
- val lock = CarbonLockFactory.getCarbonLockObj(
- carbonTable.getAbsoluteTableIdentifier,
- LockUsage.COMPACTION_LOCK)
-
- if (lock.lockWithRetries()) {
- LOGGER.info("Acquired the compaction lock for table" +
- s" ${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
- try {
- if (compactionType == CompactionType.SEGMENT_INDEX) {
- // Just launch job to merge index and return
- CommonUtil.mergeIndexFiles(sqlContext.sparkContext,
- CarbonDataMergerUtil.getValidSegmentList(
- carbonTable.getAbsoluteTableIdentifier).asScala,
- carbonLoadModel.getTablePath,
- carbonTable, true)
- lock.unlock()
- return
- }
- CarbonDataRDDFactory.startCompactionThreads(
- sqlContext,
- carbonLoadModel,
- storeLocation,
- compactionModel,
- lock
- )
- } catch {
- case e: Exception =>
- LOGGER.error(s"Exception in start compaction thread. ${ e.getMessage }")
- lock.unlock()
- throw e
- }
- } else {
- LOGGER.audit("Not able to acquire the compaction lock for table " +
- s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
- LOGGER.error(s"Not able to acquire the compaction lock for table" +
- s" ${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
- CarbonException.analysisException(
- "Table is already locked for compaction. Please try after some time.")
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
new file mode 100644
index 0000000..cbc1412
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonAlterTableCompactionCommand.scala
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command.management
+
+import scala.collection.JavaConverters._
+
+import org.apache.spark.sql.{CarbonEnv, Row, SparkSession, SQLContext}
+import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+import org.apache.spark.sql.execution.command.{AlterTableModel, CompactionModel, DataCommand, DataProcessOperation, RunnableCommand}
+import org.apache.spark.sql.hive.CarbonRelation
+import org.apache.spark.sql.util.CarbonException
+
+import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
+import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, TableInfo}
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
+import org.apache.carbondata.processing.merger.{CarbonDataMergerUtil, CompactionType}
+import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory
+import org.apache.carbondata.spark.util.CommonUtil
+import org.apache.carbondata.streaming.StreamHandoffRDD
+
+/**
+ * Command for the compaction in alter table command
+ */
+case class CarbonAlterTableCompactionCommand(
+ alterTableModel: AlterTableModel,
+ tableInfoOp: Option[TableInfo] = None)
+ extends DataCommand {
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ val LOGGER: LogService =
+ LogServiceFactory.getLogService(this.getClass.getName)
+ val tableName = alterTableModel.tableName.toLowerCase
+ val databaseName = alterTableModel.dbName.getOrElse(sparkSession.catalog.currentDatabase)
+
+ val table = if (tableInfoOp.isDefined) {
+ val tableInfo = tableInfoOp.get
+ // To DO: CarbonEnv.updateStorePath
+ CarbonTable.buildFromTableInfo(tableInfo)
+ } else {
+ val relation =
+ CarbonEnv.getInstance(sparkSession).carbonMetastore
+ .lookupRelation(Option(databaseName), tableName)(sparkSession)
+ .asInstanceOf[CarbonRelation]
+ if (relation == null) {
+ throw new NoSuchTableException(databaseName, tableName)
+ }
+ if (null == relation.carbonTable) {
+ LOGGER.error(s"alter table failed. table not found: $databaseName.$tableName")
+ throw new NoSuchTableException(databaseName, tableName)
+ }
+ relation.carbonTable
+ }
+
+ val carbonLoadModel = new CarbonLoadModel()
+ carbonLoadModel.setTableName(table.getTableName)
+ val dataLoadSchema = new CarbonDataLoadSchema(table)
+ // Need to fill dimension relation
+ carbonLoadModel.setCarbonDataLoadSchema(dataLoadSchema)
+ carbonLoadModel.setTableName(table.getTableName)
+ carbonLoadModel.setDatabaseName(table.getDatabaseName)
+ carbonLoadModel.setTablePath(table.getTablePath)
+
+ var storeLocation = CarbonProperties.getInstance.getProperty(
+ CarbonCommonConstants.STORE_LOCATION_TEMP_PATH,
+ System.getProperty("java.io.tmpdir"))
+ storeLocation = storeLocation + "/carbonstore/" + System.nanoTime()
+ try {
+ alterTableForCompaction(
+ sparkSession.sqlContext,
+ alterTableModel,
+ carbonLoadModel,
+ storeLocation)
+ } catch {
+ case e: Exception =>
+ if (null != e.getMessage) {
+ CarbonException.analysisException(
+ s"Compaction failed. Please check logs for more info. ${ e.getMessage }")
+ } else {
+ CarbonException.analysisException(
+ "Exception in compaction. Please check logs for more info.")
+ }
+ }
+ Seq.empty
+ }
+
+ private def alterTableForCompaction(sqlContext: SQLContext,
+ alterTableModel: AlterTableModel,
+ carbonLoadModel: CarbonLoadModel,
+ storeLocation: String): Unit = {
+ val LOGGER: LogService = LogServiceFactory.getLogService(this.getClass.getName)
+ val compactionType = CompactionType.valueOf(alterTableModel.compactionType.toUpperCase)
+ val compactionSize: Long = CarbonDataMergerUtil.getCompactionSize(compactionType)
+ if (CompactionType.IUD_UPDDEL_DELTA == compactionType) {
+ if (alterTableModel.segmentUpdateStatusManager.isDefined) {
+ carbonLoadModel.setSegmentUpdateStatusManager(
+ alterTableModel.segmentUpdateStatusManager.get)
+ carbonLoadModel.setLoadMetadataDetails(
+ alterTableModel.segmentUpdateStatusManager.get.getLoadMetadataDetails.toList.asJava)
+ }
+ }
+
+ LOGGER.audit(s"Compaction request received for table " +
+ s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
+
+ if (null == carbonLoadModel.getLoadMetadataDetails) {
+ CommonUtil.readLoadMetadataDetails(carbonLoadModel)
+ }
+
+ if (compactionType == CompactionType.STREAMING) {
+ StreamHandoffRDD.startStreamingHandoffThread(
+ carbonLoadModel,
+ sqlContext,
+ storeLocation)
+ return
+ }
+
+ // reading the start time of data load.
+ val loadStartTime : Long =
+ if (alterTableModel.factTimeStamp.isEmpty) {
+ CarbonUpdateUtil.readCurrentTime
+ } else {
+ alterTableModel.factTimeStamp.get
+ }
+ carbonLoadModel.setFactTimeStamp(loadStartTime)
+
+ val isCompactionTriggerByDDl = true
+ val compactionModel = CompactionModel(compactionSize,
+ compactionType,
+ carbonTable,
+ isCompactionTriggerByDDl
+ )
+
+ val isConcurrentCompactionAllowed = CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.ENABLE_CONCURRENT_COMPACTION,
+ CarbonCommonConstants.DEFAULT_ENABLE_CONCURRENT_COMPACTION
+ )
+ .equalsIgnoreCase("true")
+
+ // if system level compaction is enabled then only one compaction can run in the system
+ // if any other request comes at this time then it will create a compaction request file.
+ // so that this will be taken up by the compaction process which is executing.
+ if (!isConcurrentCompactionAllowed) {
+ LOGGER.info("System level compaction lock is enabled.")
+ CarbonDataRDDFactory.handleCompactionForSystemLocking(
+ sqlContext,
+ carbonLoadModel,
+ storeLocation,
+ compactionType,
+ carbonTable,
+ compactionModel
+ )
+ } else {
+ // normal flow of compaction
+ val lock = CarbonLockFactory.getCarbonLockObj(
+ carbonTable.getAbsoluteTableIdentifier,
+ LockUsage.COMPACTION_LOCK)
+
+ if (lock.lockWithRetries()) {
+ LOGGER.info("Acquired the compaction lock for table" +
+ s" ${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ try {
+ if (compactionType == CompactionType.SEGMENT_INDEX) {
+ // Just launch job to merge index and return
+ CommonUtil.mergeIndexFiles(sqlContext.sparkContext,
+ CarbonDataMergerUtil.getValidSegmentList(
+ carbonTable.getAbsoluteTableIdentifier).asScala,
+ carbonLoadModel.getTablePath,
+ carbonTable, true)
+ lock.unlock()
+ return
+ }
+ CarbonDataRDDFactory.startCompactionThreads(
+ sqlContext,
+ carbonLoadModel,
+ storeLocation,
+ compactionModel,
+ lock
+ )
+ } catch {
+ case e: Exception =>
+ LOGGER.error(s"Exception in start compaction thread. ${ e.getMessage }")
+ lock.unlock()
+ throw e
+ }
+ } else {
+ LOGGER.audit("Not able to acquire the compaction lock for table " +
+ s"${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ LOGGER.error(s"Not able to acquire the compaction lock for table" +
+ s" ${ carbonLoadModel.getDatabaseName }.${ carbonLoadModel.getTableName }")
+ CarbonException.analysisException(
+ "Table is already locked for compaction. Please try after some time.")
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonCleanFilesCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonCleanFilesCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonCleanFilesCommand.scala
new file mode 100644
index 0000000..1bb9cfc
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonCleanFilesCommand.scala
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command.management
+
+import org.apache.spark.sql.{CarbonEnv, GetDB, Row, SparkSession}
+import org.apache.spark.sql.execution.command.{Checker, DataCommand, DataProcessOperation, RunnableCommand}
+
+import org.apache.carbondata.api.CarbonStore
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.CarbonProperties
+import org.apache.carbondata.events.{CleanFilesPostEvent, CleanFilesPreEvent, OperationContext, OperationListenerBus}
+import org.apache.carbondata.spark.util.CommonUtil
+
+/**
+ * Clean data in table
+ * If table name is specified and forceTableClean is false, it will clean garbage
+ * segment (MARKED_FOR_DELETE state).
+ * If table name is specified and forceTableClean is true, it will delete all data
+ * in the table.
+ * If table name is not provided, it will clean garbage segment in all tables.
+ */
+case class CarbonCleanFilesCommand(
+ databaseNameOp: Option[String],
+ tableName: Option[String],
+ forceTableClean: Boolean = false)
+ extends DataCommand {
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName.get)(sparkSession)
+ val operationContext = new OperationContext
+ val cleanFilesPreEvent: CleanFilesPreEvent =
+ CleanFilesPreEvent(carbonTable,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(cleanFilesPreEvent, operationContext)
+ if (tableName.isDefined) {
+ Checker.validateTableExists(databaseNameOp, tableName.get, sparkSession)
+ if (forceTableClean) {
+ deleteAllData(sparkSession, databaseNameOp, tableName.get)
+ } else {
+ cleanGarbageData(sparkSession, databaseNameOp, tableName.get)
+ }
+ } else {
+ cleanGarbageDataInAllTables(sparkSession)
+ }
+ val cleanFilesPostEvent: CleanFilesPostEvent =
+ CleanFilesPostEvent(carbonTable,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(cleanFilesPreEvent)
+ Seq.empty
+ }
+
+ private def deleteAllData(sparkSession: SparkSession,
+ databaseNameOp: Option[String], tableName: String): Unit = {
+ val dbName = GetDB.getDatabaseName(databaseNameOp, sparkSession)
+ val databaseLocation = GetDB.getDatabaseLocation(dbName, sparkSession,
+ CarbonProperties.getStorePath)
+ CarbonStore.cleanFiles(
+ dbName,
+ tableName,
+ databaseLocation,
+ null,
+ forceTableClean)
+ }
+
+ private def cleanGarbageData(sparkSession: SparkSession,
+ databaseNameOp: Option[String], tableName: String): Unit = {
+ val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
+ val cleanFilesPreEvent: CleanFilesPreEvent =
+ CleanFilesPreEvent(carbonTable,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(cleanFilesPreEvent)
+
+ CarbonStore.cleanFiles(
+ GetDB.getDatabaseName(databaseNameOp, sparkSession),
+ tableName,
+ CarbonProperties.getStorePath,
+ carbonTable,
+ forceTableClean)
+
+ val cleanFilesPostEvent: CleanFilesPostEvent =
+ CleanFilesPostEvent(carbonTable,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(cleanFilesPreEvent)
+ }
+
+ private def cleanGarbageDataInAllTables(sparkSession: SparkSession): Unit = {
+ // Waiting to implement
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByIdCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByIdCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByIdCommand.scala
new file mode 100644
index 0000000..06eb657
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByIdCommand.scala
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command.management
+
+import org.apache.spark.sql.{CarbonEnv, GetDB, Row, SparkSession}
+import org.apache.spark.sql.execution.command.{Checker, DataCommand}
+
+import org.apache.carbondata.api.CarbonStore
+import org.apache.carbondata.events.{DeleteSegmentByIdPostEvent, DeleteSegmentByIdPreEvent, OperationContext, OperationListenerBus}
+
+case class CarbonDeleteLoadByIdCommand(
+ loadIds: Seq[String],
+ databaseNameOp: Option[String],
+ tableName: String)
+ extends DataCommand {
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ Checker.validateTableExists(databaseNameOp, tableName, sparkSession)
+ val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
+ val operationContext = new OperationContext
+
+ val deleteSegmentByIdPreEvent: DeleteSegmentByIdPreEvent =
+ DeleteSegmentByIdPreEvent(carbonTable,
+ loadIds,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(deleteSegmentByIdPreEvent, operationContext)
+
+ CarbonStore.deleteLoadById(
+ loadIds,
+ GetDB.getDatabaseName(databaseNameOp, sparkSession),
+ tableName,
+ carbonTable
+ )
+
+ val deleteSegmentPostEvent: DeleteSegmentByIdPostEvent =
+ DeleteSegmentByIdPostEvent(carbonTable,
+ loadIds,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(deleteSegmentPostEvent, operationContext)
+ Seq.empty
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByLoadDateCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByLoadDateCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByLoadDateCommand.scala
new file mode 100644
index 0000000..dbfb030
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonDeleteLoadByLoadDateCommand.scala
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command.management
+
+import org.apache.spark.sql.{CarbonEnv, GetDB, Row, SparkSession}
+import org.apache.spark.sql.execution.command.{Checker, DataCommand}
+
+import org.apache.carbondata.api.CarbonStore
+import org.apache.carbondata.events.{DeleteSegmentByDatePostEvent, DeleteSegmentByDatePreEvent, OperationContext, OperationListenerBus}
+
+case class CarbonDeleteLoadByLoadDateCommand(
+ databaseNameOp: Option[String],
+ tableName: String,
+ dateField: String,
+ loadDate: String)
+ extends DataCommand {
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ Checker.validateTableExists(databaseNameOp, tableName, sparkSession)
+ val carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
+ val operationContext = new OperationContext
+ val deleteSegmentByDatePreEvent: DeleteSegmentByDatePreEvent =
+ DeleteSegmentByDatePreEvent(carbonTable,
+ loadDate,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(deleteSegmentByDatePreEvent, operationContext)
+
+ CarbonStore.deleteLoadByDate(
+ loadDate,
+ GetDB.getDatabaseName(databaseNameOp, sparkSession),
+ tableName,
+ carbonTable)
+ val deleteSegmentPostEvent: DeleteSegmentByDatePostEvent =
+ DeleteSegmentByDatePostEvent(carbonTable,
+ loadDate,
+ sparkSession)
+ OperationListenerBus.getInstance.fireEvent(deleteSegmentPostEvent, operationContext)
+
+ Seq.empty
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/90fb6baf/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
new file mode 100644
index 0000000..ef16955
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CarbonInsertIntoCommand.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.command.management
+
+import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, Dataset, Row, SparkSession}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.execution.command.DataCommand
+
+import org.apache.carbondata.spark.util.CarbonSparkUtil
+
+case class CarbonInsertIntoCommand(
+ relation: CarbonDatasourceHadoopRelation,
+ child: LogicalPlan,
+ overwrite: Boolean)
+ extends DataCommand {
+
+ override def processData(sparkSession: SparkSession): Seq[Row] = {
+ val df = Dataset.ofRows(sparkSession, child)
+ val header = relation.tableSchema.get.fields.map(_.name).mkString(",")
+ val load = CarbonLoadDataCommand(
+ Some(relation.carbonRelation.databaseName),
+ relation.carbonRelation.tableName,
+ null,
+ Seq(),
+ scala.collection.immutable.Map("fileheader" -> header),
+ overwrite,
+ null,
+ Some(df)).run(sparkSession)
+ // updating relation metadata. This is in case of auto detect high cardinality
+ relation.carbonRelation.metaData =
+ CarbonSparkUtil.createSparkMeta(relation.carbonRelation.carbonTable)
+ load
+ }
+}