You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2017/11/25 08:39:45 UTC
carbondata git commit: [CARBONDATA-1778] Support clean data for all
Repository: carbondata
Updated Branches:
refs/heads/master 85f039203 -> 520ee50a7
[CARBONDATA-1778] Support clean data for all
Modification reasons:
Now Carbon only support clean garbage segments for specified table.
Carbon should provide the ability to clean all garbage segments without specified the database name and table name.
This closes #1537
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/520ee50a
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/520ee50a
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/520ee50a
Branch: refs/heads/master
Commit: 520ee50a759aacdf9dc4864a07c957b0eccdfcf8
Parents: 85f0392
Author: chenerlu <ch...@huawei.com>
Authored: Mon Nov 20 17:01:42 2017 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Sat Nov 25 16:39:13 2017 +0800
----------------------------------------------------------------------
.../command/management/CleanFilesCommand.scala | 89 +++++++++++++-------
.../sql/parser/CarbonSpark2SqlParser.scala | 2 +-
.../AlterTableValidationTestCase.scala | 3 +
3 files changed, 62 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/520ee50a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CleanFilesCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CleanFilesCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CleanFilesCommand.scala
index 58e33b7..7625579 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CleanFilesCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/management/CleanFilesCommand.scala
@@ -19,15 +19,25 @@ 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, DataProcessCommand, RunnableCommand}
-import org.apache.spark.sql.hive.CarbonRelation
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 CleanFilesCommand(
databaseNameOp: Option[String],
- tableName: String, forceTableClean: Boolean = false)
+ tableName: Option[String],
+ forceTableClean: Boolean = false)
extends RunnableCommand with DataProcessCommand {
override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -35,37 +45,54 @@ case class CleanFilesCommand(
}
override def processData(sparkSession: SparkSession): Seq[Row] = {
- Checker.validateTableExists(databaseNameOp, tableName, sparkSession)
- if (forceTableClean) {
- val dbName = GetDB.getDatabaseName(databaseNameOp, sparkSession)
- val databaseLocation = GetDB.getDatabaseLocation(dbName, sparkSession,
- CarbonProperties.getStorePath)
- // TODO: TAABLEPATH
- CarbonStore.cleanFiles(
- dbName,
- tableName,
- databaseLocation,
- null,
- forceTableClean)
+ if (tableName.isDefined) {
+ Checker.validateTableExists(databaseNameOp, tableName.get, sparkSession)
+ if (forceTableClean) {
+ deleteAllData(sparkSession, databaseNameOp, tableName.get)
+ } else {
+ cleanGarbageData(sparkSession, databaseNameOp, tableName.get)
+ }
} else {
- 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)
+ cleanGarbageDataInAllTables(sparkSession)
}
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/520ee50a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index c18f4e3..0187489 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -332,7 +332,7 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
protected lazy val cleanFiles: Parser[LogicalPlan] =
CLEAN ~> FILES ~> FOR ~> TABLE ~> (ident <~ ".").? ~ ident <~ opt(";") ^^ {
case databaseName ~ tableName =>
- CleanFilesCommand(convertDbNameToLowerCase(databaseName), tableName.toLowerCase())
+ CleanFilesCommand(convertDbNameToLowerCase(databaseName), Option(tableName.toLowerCase()))
}
protected lazy val explainPlan: Parser[LogicalPlan] =
http://git-wip-us.apache.org/repos/asf/carbondata/blob/520ee50a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
index 3b6b85d..638a6e8 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
@@ -504,6 +504,9 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
assert(intercept[RuntimeException] {
sql("alter table preaggmain rename to preaggmain_new")
}.getMessage.contains("Rename operation is not supported for table with pre-aggregate tables"))
+ sql("drop table if exists preaggMain")
+ sql("drop table if exists preaggmain_new")
+ sql("drop table if exists preaggMain_preagg1")
}
override def afterAll {